312 lines
9.0 KiB
Python
312 lines
9.0 KiB
Python
"""
|
|
ASGI Application Mount Module
|
|
|
|
This module provides functionality to mount external ASGI/WSGI applications
|
|
(FastAPI, Flask, Django, etc.) at specified paths within PyServe.
|
|
"""
|
|
|
|
import importlib
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional, Callable, cast
|
|
from starlette.types import ASGIApp, Receive, Scope, Send
|
|
|
|
from .logging_utils import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class ASGIAppLoader:
|
|
def __init__(self) -> None:
|
|
self._apps: Dict[str, ASGIApp] = {}
|
|
self._wsgi_adapters: Dict[str, ASGIApp] = {}
|
|
|
|
def load_app(
|
|
self,
|
|
app_path: str,
|
|
app_type: str = "asgi",
|
|
module_path: Optional[str] = None,
|
|
factory: bool = False,
|
|
factory_args: Optional[Dict[str, Any]] = None,
|
|
) -> Optional[ASGIApp]:
|
|
try:
|
|
if module_path:
|
|
module_dir = Path(module_path).resolve()
|
|
if str(module_dir) not in sys.path:
|
|
sys.path.insert(0, str(module_dir))
|
|
logger.debug(f"Added {module_dir} to sys.path")
|
|
|
|
if ":" in app_path:
|
|
module_name, attr_name = app_path.rsplit(":", 1)
|
|
else:
|
|
module_name = app_path
|
|
attr_name = "app"
|
|
|
|
module = importlib.import_module(module_name)
|
|
|
|
app_or_factory = getattr(module, attr_name)
|
|
|
|
if factory:
|
|
factory_args = factory_args or {}
|
|
app = app_or_factory(**factory_args)
|
|
logger.info(f"Created app from factory: {app_path}")
|
|
else:
|
|
app = app_or_factory
|
|
logger.info(f"Loaded app: {app_path}")
|
|
|
|
if app_type == "wsgi":
|
|
app = self._wrap_wsgi(app)
|
|
logger.info(f"Wrapped WSGI app: {app_path}")
|
|
|
|
self._apps[app_path] = app
|
|
return cast(ASGIApp, app)
|
|
|
|
except ImportError as e:
|
|
logger.error(f"Failed to import application {app_path}: {e}")
|
|
return None
|
|
except AttributeError as e:
|
|
logger.error(f"Failed to get attribute from {app_path}: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Failed to load application {app_path}: {e}")
|
|
return None
|
|
|
|
def _wrap_wsgi(self, wsgi_app: Callable) -> ASGIApp:
|
|
try:
|
|
from a2wsgi import WSGIMiddleware
|
|
return cast(ASGIApp, WSGIMiddleware(wsgi_app))
|
|
except ImportError:
|
|
logger.warning("a2wsgi not installed, trying asgiref")
|
|
try:
|
|
from asgiref.wsgi import WsgiToAsgi
|
|
return cast(ASGIApp, WsgiToAsgi(wsgi_app))
|
|
except ImportError:
|
|
logger.error(
|
|
"Neither a2wsgi nor asgiref installed. "
|
|
"Install with: pip install a2wsgi or pip install asgiref"
|
|
)
|
|
raise ImportError(
|
|
"WSGI adapter not available. Install a2wsgi or asgiref."
|
|
)
|
|
|
|
def get_app(self, app_path: str) -> Optional[ASGIApp]:
|
|
return self._apps.get(app_path)
|
|
|
|
def reload_app(self, app_path: str, **kwargs: Any) -> Optional[ASGIApp]:
|
|
if app_path in self._apps:
|
|
del self._apps[app_path]
|
|
|
|
if ":" in app_path:
|
|
module_name, _ = app_path.rsplit(":", 1)
|
|
else:
|
|
module_name = app_path
|
|
|
|
if module_name in sys.modules:
|
|
importlib.reload(sys.modules[module_name])
|
|
|
|
return self.load_app(app_path, **kwargs)
|
|
|
|
|
|
class MountedApp:
|
|
def __init__(
|
|
self,
|
|
path: str,
|
|
app: ASGIApp,
|
|
name: str = "",
|
|
strip_path: bool = True,
|
|
):
|
|
self.path = path.rstrip("/")
|
|
self.app = app
|
|
self.name = name or path
|
|
self.strip_path = strip_path
|
|
|
|
def matches(self, request_path: str) -> bool:
|
|
if self.path == "":
|
|
return True
|
|
return request_path == self.path or request_path.startswith(f"{self.path}/")
|
|
|
|
def get_modified_path(self, original_path: str) -> str:
|
|
if not self.strip_path:
|
|
return original_path
|
|
|
|
if self.path == "":
|
|
return original_path
|
|
|
|
new_path = original_path[len(self.path):]
|
|
return new_path if new_path else "/"
|
|
|
|
|
|
class ASGIMountManager:
|
|
def __init__(self) -> None:
|
|
self._mounts: list[MountedApp] = []
|
|
self._loader = ASGIAppLoader()
|
|
|
|
def mount(
|
|
self,
|
|
path: str,
|
|
app: Optional[ASGIApp] = None,
|
|
app_path: Optional[str] = None,
|
|
app_type: str = "asgi",
|
|
module_path: Optional[str] = None,
|
|
factory: bool = False,
|
|
factory_args: Optional[Dict[str, Any]] = None,
|
|
name: str = "",
|
|
strip_path: bool = True,
|
|
) -> bool:
|
|
if app is None and app_path is None:
|
|
logger.error("Either 'app' or 'app_path' must be provided")
|
|
return False
|
|
|
|
if app is None:
|
|
app = self._loader.load_app(
|
|
app_path=app_path, # type: ignore
|
|
app_type=app_type,
|
|
module_path=module_path,
|
|
factory=factory,
|
|
factory_args=factory_args,
|
|
)
|
|
if app is None:
|
|
return False
|
|
|
|
mounted = MountedApp(
|
|
path=path,
|
|
app=app,
|
|
name=name or app_path or "unnamed",
|
|
strip_path=strip_path,
|
|
)
|
|
|
|
self._mounts.append(mounted)
|
|
self._mounts.sort(key=lambda m: len(m.path), reverse=True)
|
|
|
|
logger.info(f"Mounted application '{mounted.name}' at path '{path}'")
|
|
return True
|
|
|
|
def unmount(self, path: str) -> bool:
|
|
for i, mount in enumerate(self._mounts):
|
|
if mount.path == path.rstrip("/"):
|
|
del self._mounts[i]
|
|
logger.info(f"Unmounted application at path '{path}'")
|
|
return True
|
|
return False
|
|
|
|
def get_mount(self, request_path: str) -> Optional[MountedApp]:
|
|
for mount in self._mounts:
|
|
if mount.matches(request_path):
|
|
return mount
|
|
return None
|
|
|
|
async def handle_request(
|
|
self,
|
|
scope: Scope,
|
|
receive: Receive,
|
|
send: Send,
|
|
) -> bool:
|
|
if scope["type"] != "http":
|
|
return False
|
|
|
|
path = scope.get("path", "/")
|
|
mount = self.get_mount(path)
|
|
|
|
if mount is None:
|
|
return False
|
|
|
|
modified_scope = dict(scope)
|
|
if mount.strip_path:
|
|
modified_scope["path"] = mount.get_modified_path(path)
|
|
modified_scope["root_path"] = scope.get("root_path", "") + mount.path
|
|
|
|
logger.debug(
|
|
f"Routing request to mounted app '{mount.name}': "
|
|
f"{path} -> {modified_scope['path']}"
|
|
)
|
|
|
|
try:
|
|
await mount.app(modified_scope, receive, send)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Error in mounted app '{mount.name}': {e}")
|
|
raise
|
|
|
|
@property
|
|
def mounts(self) -> list[MountedApp]:
|
|
return self._mounts.copy()
|
|
|
|
def list_mounts(self) -> list[Dict[str, Any]]:
|
|
return [
|
|
{
|
|
"path": mount.path,
|
|
"name": mount.name,
|
|
"strip_path": mount.strip_path,
|
|
}
|
|
for mount in self._mounts
|
|
]
|
|
|
|
|
|
def create_fastapi_app(
|
|
app_path: str,
|
|
module_path: Optional[str] = None,
|
|
factory: bool = False,
|
|
factory_args: Optional[Dict[str, Any]] = None,
|
|
) -> Optional[ASGIApp]:
|
|
loader = ASGIAppLoader()
|
|
return loader.load_app(
|
|
app_path=app_path,
|
|
app_type="asgi",
|
|
module_path=module_path,
|
|
factory=factory,
|
|
factory_args=factory_args,
|
|
)
|
|
|
|
|
|
def create_flask_app(
|
|
app_path: str,
|
|
module_path: Optional[str] = None,
|
|
factory: bool = False,
|
|
factory_args: Optional[Dict[str, Any]] = None,
|
|
) -> Optional[ASGIApp]:
|
|
loader = ASGIAppLoader()
|
|
return loader.load_app(
|
|
app_path=app_path,
|
|
app_type="wsgi",
|
|
module_path=module_path,
|
|
factory=factory,
|
|
factory_args=factory_args,
|
|
)
|
|
|
|
|
|
def create_django_app(
|
|
settings_module: str,
|
|
module_path: Optional[str] = None,
|
|
) -> Optional[ASGIApp]:
|
|
import os
|
|
|
|
if module_path:
|
|
module_dir = Path(module_path).resolve()
|
|
if str(module_dir) not in sys.path:
|
|
sys.path.insert(0, str(module_dir))
|
|
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", settings_module)
|
|
|
|
try:
|
|
from django.core.asgi import get_asgi_application # type: ignore[import-untyped]
|
|
return cast(ASGIApp, get_asgi_application())
|
|
except ImportError as e:
|
|
logger.error(f"Failed to load Django application: {e}")
|
|
return None
|
|
|
|
|
|
def create_starlette_app(
|
|
app_path: str,
|
|
module_path: Optional[str] = None,
|
|
factory: bool = False,
|
|
factory_args: Optional[Dict[str, Any]] = None,
|
|
) -> Optional[ASGIApp]:
|
|
loader = ASGIAppLoader()
|
|
return loader.load_app(
|
|
app_path=app_path,
|
|
app_type="asgi",
|
|
module_path=module_path,
|
|
factory=factory,
|
|
factory_args=factory_args,
|
|
)
|