67 lines
1.9 KiB
Python
67 lines
1.9 KiB
Python
"""
|
|
WSGI Wrapper Module for Process Orchestration.
|
|
|
|
This module provides a wrapper that allows WSGI applications to be run
|
|
via uvicorn by wrapping them with a2wsgi.
|
|
|
|
The WSGI app path is passed via environment variables:
|
|
- PYSERVE_WSGI_APP: The app path (e.g., "myapp:app" or "myapp.main:create_app")
|
|
- PYSERVE_WSGI_FACTORY: "1" if the app path points to a factory function
|
|
"""
|
|
|
|
import importlib
|
|
import os
|
|
from typing import Any, Callable
|
|
|
|
try:
|
|
from a2wsgi import WSGIMiddleware
|
|
|
|
WSGI_ADAPTER = "a2wsgi"
|
|
except ImportError:
|
|
try:
|
|
from asgiref.wsgi import WsgiToAsgi as WSGIMiddleware # type: ignore
|
|
|
|
WSGI_ADAPTER = "asgiref"
|
|
except ImportError:
|
|
WSGIMiddleware = None # type: ignore
|
|
WSGI_ADAPTER = None
|
|
|
|
|
|
def _load_wsgi_app() -> Callable:
|
|
app_path = os.environ.get("PYSERVE_WSGI_APP")
|
|
is_factory = os.environ.get("PYSERVE_WSGI_FACTORY", "0") == "1"
|
|
|
|
if not app_path:
|
|
raise RuntimeError("PYSERVE_WSGI_APP environment variable not set. " "This module should only be used by PyServe process orchestration.")
|
|
|
|
if ":" in app_path:
|
|
module_name, attr_name = app_path.rsplit(":", 1)
|
|
else:
|
|
module_name = app_path
|
|
attr_name = "app"
|
|
|
|
try:
|
|
module = importlib.import_module(module_name)
|
|
except ImportError as e:
|
|
raise RuntimeError(f"Failed to import WSGI module '{module_name}': {e}")
|
|
|
|
try:
|
|
app_or_factory = getattr(module, attr_name)
|
|
except AttributeError:
|
|
raise RuntimeError(f"Module '{module_name}' has no attribute '{attr_name}'")
|
|
|
|
if is_factory:
|
|
return app_or_factory()
|
|
return app_or_factory
|
|
|
|
|
|
def _create_asgi_app() -> Any:
|
|
if WSGIMiddleware is None:
|
|
raise RuntimeError("No WSGI adapter available. " "Install a2wsgi (recommended) or asgiref: pip install a2wsgi")
|
|
|
|
wsgi_app = _load_wsgi_app()
|
|
return WSGIMiddleware(wsgi_app)
|
|
|
|
|
|
app = _create_asgi_app()
|