Coverage for src/lexigram/web/di/route_setup.py: 21%
71 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Route registration helper for the Lexigram web layer."""
3from __future__ import annotations
5from typing import TYPE_CHECKING, Any, cast
7from lexigram.contracts.exceptions.container import UnresolvableDependencyError
8from lexigram.logging import get_logger
10if TYPE_CHECKING:
11 from starlette.applications import Starlette
13 from lexigram.contracts.core.di import ContainerResolverProtocol
14 from lexigram.web.config import WebConfig, WebProviderConfig
15 from lexigram.web.di.provider import WebProvider
16 from lexigram.web.routing.manager import WebRouterManager
18logger = get_logger(__name__)
20__all__ = ["RouteSetup"]
23class RouteSetup:
24 """Encapsulates route and mount registration for the Lexigram web layer.
26 Extracted from ``WebProvider`` to keep the provider focused on DI lifecycle
27 (register/boot/shutdown) rather than application routing configuration.
29 Args:
30 web_config: Web configuration; drives debug routes, static files, etc.
31 provider_config: Provider-level configuration.
32 router_manager: The router manager that discovers and mounts controllers.
33 """
35 def __init__(
36 self,
37 web_config: WebConfig,
38 provider_config: WebProviderConfig,
39 router_manager: WebRouterManager,
40 ) -> None:
41 self._web_config = web_config
42 self._provider_config = provider_config
43 self._router_manager = router_manager
45 async def configure(
46 self,
47 app: Starlette,
48 container: ContainerResolverProtocol,
49 provider_context: Any = None,
50 ) -> None:
51 """Register all routes and ASGI mounts on the application.
53 Phases:
55 1. Debug routes (guarded by ``web.enable_debug_routes_env_gate`` config).
56 2. Prometheus ``/metrics`` ASGI app (when ``MonitoringProvider`` is active).
57 3. Canonical health endpoint (before contributors so extension
58 convenience routes cannot shadow the framework contract).
59 4. Web contributor mounts (admin panels, sub-apps, etc.).
60 5. Static file directory (when ``web_config.static`` is configured).
61 6. Main controller discovery via :class:`~lexigram.web.routing.manager.WebRouterManager`.
62 7. Register the Starlette app as the ASGI handler on the ``Application`` instance.
64 Args:
65 app: The Starlette application to configure.
66 container: The resolved DI container.
67 provider_context: Optional reference to the parent ``WebProvider``
68 instance; forwarded to integrations that need provider-level
69 state (e.g. ``DebugIntegration`` for Redis client caching).
70 """
71 await self._register_debug_routes(app, container, provider_context)
72 await self._mount_metrics(app, container)
73 self._register_health_route(app)
74 await self._mount_contributors(app, container, provider_context)
75 self._mount_static(app)
76 await self._router_manager.register_routes(app, container)
77 await self._register_asgi_handler(app, container)
79 def _register_health_route(self, app: Starlette) -> None:
80 """Register the canonical health endpoint before contributor mounts.
82 Contributors (e.g. the relay gateway) may offer their own ``/health``
83 convenience route with dedup logic; registering ours first guarantees
84 the framework's 200/207/503 health contract wins regardless of which
85 extensions are installed. Later duplicate registrations are no-ops.
86 """
87 from lexigram.web.routing.health import register_health_route
89 # app is a Starlette instance; the health helper accepts any ASGI
90 # application with a .routes attribute, so the nominal mismatch is safe.
91 register_health_route(
92 cast("WebProvider", self._router_manager.provider),
93 app, # type: ignore[arg-type] # Starlette vs Application nominal mismatch
94 )
96 # ------------------------------------------------------------------
97 # Private helpers
98 # ------------------------------------------------------------------
100 async def _register_debug_routes(
101 self,
102 app: Starlette,
103 container: ContainerResolverProtocol,
104 provider_context: Any,
105 ) -> None:
106 """Register debug routes when enabled and gate allows registration."""
107 from lexigram.web.integrations.debug import DebugIntegration
109 debug_enabled = self._web_config.debug_routes or getattr(
110 self._provider_config,
111 "debug_routes",
112 False,
113 )
114 if not debug_enabled:
115 return
117 if not getattr(self._web_config, "enable_debug_routes_env_gate", False):
118 logger.warning(
119 "debug_routes_gate_disabled",
120 message="debug_routes=True in config but "
121 "enable_debug_routes_env_gate=False. Debug routes will not be registered.",
122 )
123 return
125 await DebugIntegration.configure(app, cast("Any", container), provider_context)
127 async def _mount_metrics(
128 self,
129 app: Starlette,
130 container: ContainerResolverProtocol,
131 ) -> None:
132 """Mount Prometheus /metrics ASGI app when ``MonitoringProvider`` is active."""
133 try:
134 metrics_asgi_app: Any = await container.resolve(
135 cast("Any", "prometheus_metrics_app"),
136 bypass_visibility=True,
137 )
138 if metrics_asgi_app is not None:
139 metrics_path = getattr(self._web_config, "metrics_path", "/metrics")
140 app.mount(metrics_path, metrics_asgi_app)
141 logger.info("web.metrics_endpoint_mounted", path=metrics_path)
142 except (UnresolvableDependencyError, RuntimeError, AttributeError, TypeError):
143 pass
145 async def _mount_contributors(
146 self,
147 app: Starlette,
148 container: ContainerResolverProtocol,
149 provider_context: Any,
150 ) -> None:
151 """Mount web contributor sub-apps generically.
153 Iterates all registered web contributors and calls their mount_to_app
154 hook. Failures are isolated and logged; they do not block other
155 contributors or route setup.
157 Args:
158 app: The Starlette application.
159 container: The DI container.
160 provider_context: The WebProvider instance (for registry access).
161 """
162 if provider_context is None or not hasattr(
163 provider_context, "contributor_registry"
164 ):
165 return
167 registry = provider_context.contributor_registry
168 for contributor in registry.get_all():
169 try:
170 await contributor.mount_to_app(app, container)
171 logger.debug(
172 "web.contributor_mount_ok",
173 contributor_id=contributor.contributor_id,
174 )
175 except Exception as mount_exc:
176 logger.error(
177 "web.contributor_mount_failed",
178 contributor_id=contributor.contributor_id,
179 error=str(mount_exc),
180 exc_type=type(mount_exc).__name__,
181 )
183 def _mount_static(self, app: Starlette) -> None:
184 """Mount static file directory when configured."""
185 static_config = getattr(self._web_config, "static", None)
186 if static_config is None or not getattr(static_config, "enabled", False):
187 return
189 from starlette.staticfiles import StaticFiles
191 try:
192 app.mount(
193 static_config.prefix,
194 StaticFiles(
195 directory=static_config.directory,
196 html=getattr(static_config, "html", False),
197 ),
198 name="static",
199 )
200 logger.info(
201 "web.static_files_mounted",
202 prefix=static_config.prefix,
203 directory=static_config.directory,
204 )
205 except (OSError, RuntimeError, ValueError) as exc:
206 logger.warning(
207 "web.static_files_mount_failed",
208 directory=static_config.directory,
209 error=str(exc),
210 )
212 async def _register_asgi_handler(
213 self,
214 app: Starlette,
215 container: ContainerResolverProtocol,
216 ) -> None:
217 """Register the Starlette app as the ASGI handler on the Application."""
218 from lexigram.app.base import Application
220 lex_app = None
221 try:
222 lex_app = await container.resolve(
223 Application,
224 bypass_visibility=True,
225 )
226 except (AttributeError, UnresolvableDependencyError) as exc:
227 logger.warning("app_resolution_failed", error=str(exc))
229 if lex_app:
230 if hasattr(lex_app, "set_asgi_handler"):
231 lex_app.set_asgi_handler(app)
232 else:
233 lex_app._asgi_handler = app # type: ignore[attr-defined]