Coverage for src/lexigram/web/di/provider.py: 23%
205 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""WebProvider - Main web provider with pass-through architecture"""
3from __future__ import annotations
5from collections.abc import Callable
6from typing import Any, cast
8from starlette.applications import Starlette
9from starlette.responses import JSONResponse
11from lexigram.contracts.core import (
12 HealthCheckResult,
13 HealthStatus,
14 HookRegistryProtocol,
15 ProviderPriority,
16)
17from lexigram.contracts.core.di import (
18 ContainerRegistrarProtocol,
19 ContainerResolverProtocol,
20)
21from lexigram.contracts.exceptions.config import ConfigurationError
22from lexigram.contracts.web import WebProviderProtocol
23from lexigram.di.provider import Provider
24from lexigram.logging import get_logger
25from lexigram.web.config import WebConfig, WebProviderConfig
26from lexigram.web.di.middleware_setup import MiddlewareSetup
27from lexigram.web.di.route_setup import RouteSetup
28from lexigram.web.docs.generator import OpenAPIGenerator
29from lexigram.web.integrations.auth import AuthIntegration
30from lexigram.web.integrations.cache import CacheIntegration
31from lexigram.web.integrations.graphql import GraphQLIntegration
32from lexigram.web.integrations.rate_limit import RateLimitIntegration
33from lexigram.web.integrations.setup import lifespan
34from lexigram.web.integrations.sql import SQLIntegration
35from lexigram.web.middleware.manager import WebMiddlewareManager
36from lexigram.web.routing.controllers import Controller
37from lexigram.web.routing.manager import WebRouterManager
38from lexigram.web.routing.router import Router
40# Optional external server - imported lazily to avoid dependency issues
41# Only needed when run_server() is called
43logger = get_logger(__name__)
46class WebProvider(Provider):
47 """Web provider with native Starlette integration and pass-through architecture.
49 **Recommended usage** — let the orchestrator inject configuration via
50 ``application.yaml`` (uses ``config_key = "web"`` and ``config_model = WebConfig``)::
52 app.add_provider(WebProvider())
54 **Config-first factory** — construct from an explicit ``WebConfig``::
56 from lexigram.web import WebProvider, WebConfig
58 app.add_provider(WebProvider.from_config(WebConfig(debug=True, ...)))
60 **Advanced** — pass individual components explicitly (e.g. for tests)::
62 app.add_provider(WebProvider(
63 controllers=[UsersController],
64 middleware=[AuthMiddleware],
65 web_config=WebConfig(debug=True),
66 ))
68 Note: The multi-parameter constructor is intended for advanced and test
69 scenarios. For typical applications, prefer the no-arg or ``from_config()``
70 form so that configuration is driven by ``application.yaml``.
71 """
73 name = "web"
74 priority = ProviderPriority.PRESENTATION
76 def __init__(
77 self,
78 middleware: list[Any] | None = None,
79 exception_handlers: dict[Any, Any] | None = None,
80 controllers: list[type[Controller]] | None = None,
81 web_config: WebConfig | None = None,
82 provider_config: WebProviderConfig | None = None,
83 debug_routes_auth: Callable[..., Any] | None = None,
84 ) -> None:
85 super().__init__()
86 self.middleware = middleware or []
87 self.exception_handlers = exception_handlers or {}
88 self.controllers = controllers or []
90 self.web_config = web_config or WebConfig()
91 self._explicit_web_config = web_config is not None
92 self.provider_config = provider_config or WebProviderConfig()
94 self.starlette: Starlette | None = None
95 self._hook_registry: HookRegistryProtocol | None = None
97 # Managers
98 self.middleware_manager = WebMiddlewareManager(self)
99 self.router_manager: WebRouterManager = WebRouterManager(self)
101 # Internal state for routing
102 self.router: Router | None = None # Lazy loaded if needed
103 self.openapi_generator: OpenAPIGenerator | None = None
105 # Route conflict behavior (default: do not raise on duplicate routes)
106 # Tests expect duplicate registrations to emit warnings unless explicitly configured
107 self.fail_on_route_conflict = getattr(
108 self.provider_config,
109 "fail_on_route_conflict",
110 False,
111 )
113 # Debug routes config
114 self.debug_routes_auth: Callable[..., Any] | None = debug_routes_auth
115 self._debug_redis_client: Any | None = None
117 # Extra services to register in DI (used by quickstart auto-injection)
118 self._extra_injectable_services: list[tuple[type, Any]] = []
120 # Web contributor registry for entry-point-based controller/middleware discovery
121 from lexigram.web.contributors import WebContributorRegistry
123 self._contributor_registry = WebContributorRegistry()
125 @property
126 def contributor_registry(self) -> Any:
127 """Expose contributor registry for route setup access."""
128 return self._contributor_registry
130 @classmethod
131 def from_config(cls, config: WebConfig, **context: Any) -> WebProvider:
132 """Create a WebProvider from config.
134 Context kwargs may include middleware, controllers, exception_handlers.
135 """
136 return cls(
137 web_config=config,
138 middleware=context.get("middleware"),
139 controllers=context.get("controllers"),
140 exception_handlers=context.get("exception_handlers"),
141 )
143 @classmethod
144 def auto_discover(
145 cls,
146 *packages: str,
147 web_config: WebConfig | None = None,
148 **kwargs: Any,
149 ) -> WebProvider:
150 """Create a WebProvider with controllers auto-discovered from packages.
152 Scans each package recursively for
153 :class:`~lexigram.web.routing.controllers.Controller` subclasses and
154 registers them automatically.
156 Args:
157 *packages: Dotted Python package paths to scan for controllers,
158 e.g. ``"my_app.api.controllers"``.
159 web_config: Optional web configuration. Falls back to defaults.
160 **kwargs: Extra kwargs forwarded to :class:`WebProvider.__init__`.
162 Returns:
163 A configured :class:`WebProvider` instance.
165 Example::
167 app.add_provider(WebProvider.auto_discover("my_app.api.controllers"))
168 """
169 from lexigram.web.routing.discovery import discover_controllers
171 controllers = discover_controllers(list(packages))
172 return cls(controllers=controllers, web_config=web_config, **kwargs)
174 async def register(self, container: ContainerRegistrarProtocol) -> None:
175 """Register web services in DI container"""
177 # GuardProtocol: debug routes with no protection would expose the entire DI graph.
178 # Require at least one protection mechanism: a token or an auth callback.
179 if (
180 self.web_config.debug_routes
181 and not self.web_config.debug_routes_token
182 and self.debug_routes_auth is None
183 ):
184 raise ConfigurationError(
185 "debug_routes=True requires either debug_routes_token (in WebConfig) "
186 "or a debug_routes_auth callback (in WebProvider). "
187 "Set one to protect the debug endpoint."
188 )
190 container.singleton(WebProvider, self)
191 container.singleton(WebProviderProtocol, self)
193 from lexigram.contracts.web.sse import ReactiveSseBridgeProtocol
194 from lexigram.web.transport.reactive import sse_from_stream
196 container.singleton(ReactiveSseBridgeProtocol, sse_from_stream)
198 from lexigram.primitives.context import Context, create_default_context
200 container.singleton(Context, create_default_context())
202 from lexigram.web.security.config import (
203 CORSConfig,
204 CrossOriginConfig,
205 CSPConfig,
206 CSRFConfig,
207 HSTSConfig,
208 SecurityConfig,
209 SecurityHeadersConfig,
210 )
211 from lexigram.web.security.cors.middleware import CORSMiddlewareFactory
213 container.singleton(SecurityConfig, self.web_config.security)
214 container.singleton(CORSConfig, self.web_config.cors)
215 container.singleton(CSRFConfig, self.web_config.security.csrf)
216 container.singleton(SecurityHeadersConfig, self.web_config.security.headers)
217 container.singleton(HSTSConfig, self.web_config.security.hsts)
218 container.singleton(CSPConfig, self.web_config.security.csp)
219 container.singleton(CrossOriginConfig, self.web_config.security.cross_origin)
220 container.singleton(
221 CORSMiddlewareFactory,
222 CORSMiddlewareFactory(config=self.web_config.cors),
223 )
225 # Register the global route registry so DI resolution returns the same
226 # instance that @route decorators populated at import time.
227 from lexigram.web.routing.registry import RouteRegistry, route_registry
229 container.singleton(RouteRegistry, route_registry)
231 # Register the global controller registry so DI resolution returns the same
232 # instance that @controller decorators populated at import time.
233 from lexigram.web.routing.controller_registry import (
234 ControllerRegistry,
235 controller_registry,
236 )
238 container.singleton(ControllerRegistry, controller_registry)
240 # Register FilterPipeline and InterceptorPipeline as container singletons so
241 # they can be injected rather than accessed via module-level globals.
242 from lexigram.web.filters.pipeline import FilterPipeline, filter_pipeline
244 container.singleton(FilterPipeline, filter_pipeline)
246 from lexigram.web.interceptors.pipeline import InterceptorPipeline
248 container.singleton(InterceptorPipeline, InterceptorPipeline())
250 # Register Router for dependency injection (pre-instantiated so the DI
251 # framework does not inject the FilterPipeline singleton into it and
252 # accidentally pollute the global filter pipeline with Router-local filters).
253 container.singleton(Router, Router())
255 # Register ResponseFactoryProtocol
256 from lexigram.contracts.web import ResponseFactoryProtocol
257 from lexigram.web.responses import StarletteResponseAdapter
259 container.singleton(ResponseFactoryProtocol, StarletteResponseAdapter)
261 # Register ResponseSerializer so the router can resolve it per-request
262 # without hitting an UnresolvableDependencyError.
263 from lexigram.web.serialization.serializers import ResponseSerializer
265 container.singleton(ResponseSerializer, ResponseSerializer())
267 # Register BackgroundTaskRunnerProtocol — per-resolution (transient) so each
268 # caller gets an independent task accumulator bound to its own Starlette context.
269 # Background tasks are in-process only. Durable job submission uses explicit
270 # lexigram-tasks job APIs, not this web background-runner interface.
271 from lexigram.contracts.web.protocols import BackgroundTaskRunnerProtocol
272 from lexigram.web.background.tasks import StarletteBackgroundTaskRunner
274 container.transient(
275 cast("Any", BackgroundTaskRunnerProtocol),
276 cast("Any", StarletteBackgroundTaskRunner),
277 )
279 # Note: ObjectMapperProtocol is NOT auto-registered here because
280 # lexigram-mapping is a separate extension package. Register it
281 # explicitly via MappingModule.configure() in your application setup.
283 # Register admin widget handlers (transient for scope safety)
284 from lexigram.web.admin.contributor import WebAdminContributor
285 from lexigram.web.admin.handlers.active_connections import (
286 ActiveConnectionsWidgetHandler,
287 )
288 from lexigram.web.admin.handlers.request_rate import (
289 RequestRateWidgetHandler,
290 )
291 from lexigram.web.admin.handlers.server_status import (
292 ServerStatusWidgetHandler,
293 )
295 container.transient(ServerStatusWidgetHandler, ServerStatusWidgetHandler)
296 container.transient(
297 ActiveConnectionsWidgetHandler,
298 ActiveConnectionsWidgetHandler,
299 )
300 container.transient(RequestRateWidgetHandler, RequestRateWidgetHandler)
301 container.singleton(WebAdminContributor, WebAdminContributor)
303 # Register the web contributor registry as a singleton
304 from lexigram.web.contributors import WebContributorRegistry
305 from lexigram.web.contributors import discovery as contributor_discovery
307 container.singleton(WebContributorRegistry, self._contributor_registry)
309 # Discover and merge web contributors from entry-points
310 for contributor in contributor_discovery.load_web_contributors():
311 self._contributor_registry.register(contributor)
313 # Merge contributed middleware (avoid duplicates)
314 for middleware_cls in contributor.get_middleware():
315 if middleware_cls not in self.middleware:
316 self.middleware.append(middleware_cls)
318 # Merge contributed controllers (avoid duplicates and
319 # subclass-takes-precedence — if a subclass of the contributed
320 # controller is already registered, skip the contributed one).
321 for controller_cls in contributor.get_controllers():
322 if controller_cls not in self.controllers:
323 # Skip if a registered controller is a subclass — the
324 # user-supplied override should take precedence over the
325 # framework's own controller.
326 if isinstance(controller_cls, type) and any(
327 isinstance(ec, type)
328 and ec is not controller_cls
329 and issubclass(ec, controller_cls)
330 for ec in self.controllers
331 ):
332 continue
333 self.controllers.append(controller_cls)
335 # Register controllers as singletons if they are classes
336 for controller_cls in self.controllers:
337 if isinstance(controller_cls, type):
338 container.singleton(controller_cls, controller_cls)
340 # Expose the active middleware pipeline to admin pages and tooling.
341 # The registry mirrors exactly what the app runs: the always-present
342 # DIScopeMiddleware plus contributed and user-supplied middleware.
343 from lexigram.web.middleware.base import MiddlewareRegistry
344 from lexigram.web.middleware.di_scope import DIScopeMiddleware
345 from lexigram.web.middleware.registry import (
346 MiddlewareAdapterRegistry,
347 )
349 middleware_registry = MiddlewareRegistry()
350 middleware_registry.register_middleware(DIScopeMiddleware)
351 for middleware_cls in self.middleware:
352 cls = (
353 middleware_cls
354 if isinstance(middleware_cls, type)
355 else type(middleware_cls)
356 )
357 middleware_registry.register_middleware(cls)
358 container.singleton(MiddlewareRegistry, middleware_registry)
359 container.singleton(
360 MiddlewareAdapterRegistry,
361 MiddlewareAdapterRegistry(),
362 )
364 # Auto-register user classes decorated with @singleton / @injectable.
365 # Scan loaded non-framework modules so script-mode apps work without
366 # manually calling container.singleton() for each service.
367 self._register_injectable_services(container)
369 def _register_injectable_services(
370 self, container: ContainerRegistrarProtocol
371 ) -> None:
372 """Register services explicitly provided via ``_extra_injectable_services``.
374 This replaces the former sys.modules scanning with explicit service lists,
375 making DI registration deterministic and order-independent. Services must be
376 explicitly provided to the provider at construction time.
377 """
378 from lexigram.contracts.core.scopes import ServiceScope
380 def _register_one(cls: type, scope: Any) -> None:
381 if container.has(cls):
382 return
383 if scope == ServiceScope.SINGLETON:
384 container.singleton(cls, cls)
385 elif scope == ServiceScope.SCOPED:
386 container.scoped(cls, cls)
387 else:
388 container.transient(cls, cls)
389 logger.debug("auto_registered_injectable", cls=cls.__name__, scope=scope)
391 # Register only explicitly provided services
392 for cls, scope in self._extra_injectable_services:
393 _register_one(cls, scope)
395 async def boot(self, container: ContainerResolverProtocol) -> None:
396 """Initialize the web layer in five ordered phases.
398 Phase 1 — OpenAPI generator
399 Instantiates :class:`~lexigram.web.routing.OpenAPIGenerator` with
400 title and version from :attr:`web_config`.
402 Phase 2 — Starlette application
403 Builds the native ASGI ``Starlette`` instance. The middleware stack
404 is composed **once** here; it is never rebuilt at request time.
405 Container and config are attached to ``app.state``.
407 Phase 3 — Middleware pipeline
408 Iterates registered :class:`~lexigram.web.middleware.AbstractMiddleware`
409 subclasses and wraps the Starlette app. Order follows the provider
410 registration order (outermost-first).
412 Phase 4 — Integration setup
413 Wires optional first-class integrations: authentication, rate
414 limiting, GraphQL gateway. Each integration is only activated when
415 its config key is present in the resolved container.
417 Phase 5 — Route registration
418 Discovers annotated controller methods and mounts them on the
419 Starlette router. Route-level dependencies (guards, interceptors,
420 serializers) are resolved here.
421 """
422 logger.info("Booting WebProvider")
424 # Create FilterPipeline with debug mode based on environment
425 from lexigram.logging.debug import is_debug_mode
426 from lexigram.web.filters.pipeline import FilterPipeline
428 _debug_mode = is_debug_mode() or self.web_config.server.debug
429 self.router = Router(filter_pipeline=FilterPipeline(debug=_debug_mode))
430 self._hook_registry = await container.resolve_optional(HookRegistryProtocol)
432 # 1. Initialize OpenAPI generator
433 self.openapi_generator = OpenAPIGenerator(
434 title=str(getattr(self.web_config, "openapi_title", "Lexigram API")),
435 version=str(getattr(self.web_config, "openapi_version", "0.1.0")),
436 )
438 # 2. Initialize native Starlette app
439 self.starlette = self._init_starlette(container)
440 self.starlette.state.hook_registry = self._hook_registry
442 # 3. Setup Middlewares
443 await self._setup_middleware(self.starlette, container)
445 # 4. Setup Integrations (Auth, Rate Limit, GraphQL)
446 await self._setup_integrations(self.starlette, container)
448 # 5. Register Routes
449 await self._setup_routes(self.starlette, container)
451 # 6. Boot admin contributor
452 from lexigram.web.admin.contributor import WebAdminContributor
454 contributor = await container.resolve_optional(WebAdminContributor)
455 if contributor is not None:
456 await contributor.on_admin_boot(container)
458 logger.info("Web application startup complete")
460 def _init_starlette(self, container: ContainerResolverProtocol) -> Starlette:
461 """Initialize the Starlette application instance."""
462 from lexigram.web.integrations.starlette import build_starlette_app
464 # Build initial middleware stack
465 native_middleware = self.middleware_manager.build_native_stack(container)
467 return build_starlette_app(
468 middleware=native_middleware,
469 exception_handlers=self.exception_handlers,
470 lifespan=lifespan,
471 container=container,
472 )
474 async def _setup_middleware(
475 self, app: Starlette, container: ContainerResolverProtocol
476 ) -> None:
477 """Configure application-level middlewares via :class:`MiddlewareSetup`."""
478 setup = MiddlewareSetup(self.web_config, hooks=self._hook_registry)
479 await setup.configure(app, container)
481 async def _setup_integrations(
482 self, app: Starlette, container: ContainerResolverProtocol
483 ) -> None:
484 """Configure external integrations and complex subsystems."""
485 # 1. Rate Limiting
486 await RateLimitIntegration.configure(
487 app, cast("Any", container), self.web_config
488 )
490 # 2. Authentication
491 if self.web_config.enable_auth:
492 await AuthIntegration.configure(
493 app, cast("Any", container), self.web_config
494 )
496 # 3. GraphQL - WebSocket routes (HTTP handled by web contributor)
497 await GraphQLIntegration.configure(app, container)
499 # 4. SQL Integration - attach db_pool to app.state for lifespan cleanup
500 await SQLIntegration.configure(app, container)
502 # 5. Cache Integration - attach redis_client to app.state for lifespan cleanup
503 await CacheIntegration.configure(app, container)
505 # 6. Default exception filter (handles DomainError, HTTPError, etc.)
506 from lexigram.logging.debug import is_debug_mode
507 from lexigram.web.filters import DefaultExceptionFilter
509 _debug_on = is_debug_mode() or self.web_config.server.debug
510 default_filter = DefaultExceptionFilter(debug=_debug_on)
511 if not hasattr(app.state, "exception_filters"):
512 app.state.exception_filters = []
513 app.state.exception_filters.append(default_filter)
515 # 5. Global Exception Handlers
516 from lexigram.web.exceptions import DependencyResolutionError
518 app.add_exception_handler(
519 DependencyResolutionError,
520 self._dependency_resolution_handler,
521 )
523 async def _setup_routes(
524 self, app: Starlette, container: ContainerResolverProtocol
525 ) -> None:
526 """Register application routes and mounts via :class:`RouteSetup`."""
527 setup = RouteSetup(self.web_config, self.provider_config, self.router_manager)
528 await setup.configure(app, container, provider_context=self)
530 # -- Handlers ----------------------------------------------------------
532 def _dependency_resolution_handler(
533 self, request: Any, exc: Exception
534 ) -> JSONResponse:
535 return JSONResponse(
536 {
537 "error": "dependency_resolution_error",
538 "message": str(exc),
539 "details": getattr(exc, "details", {}),
540 },
541 status_code=500,
542 )
544 async def shutdown(self) -> None:
545 """Cleanup resources created by this provider."""
546 logger.info("Shutting down WebProvider...")
547 self._debug_redis_client = None
548 self._hook_registry = None
549 self.starlette = None
550 logger.info("WebProvider shutdown complete")
552 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
553 """Check web provider health."""
554 return HealthCheckResult(
555 component="web",
556 status=HealthStatus.HEALTHY
557 if self.starlette is not None
558 else HealthStatus.UNHEALTHY,
559 details={
560 "starlette_initialized": self.starlette is not None,
561 },
562 )
564 def run_server(
565 self,
566 host: str = "127.0.0.1",
567 port: int = 8000,
568 **kwargs: Any,
569 ) -> None:
570 """Run the web application using Granian.
572 See :func:`lexigram.web.server.runner.run_server` for implementation.
573 """
574 from lexigram.web.server.runner import run_server as _run_server
576 if self.starlette is None:
577 raise RuntimeError(
578 "Starlette app not initialized. Call boot() on the provider first."
579 )
581 _run_server(self.starlette, host=host, port=port, **kwargs)
584__all__ = ["WebProvider"]