Coverage for src/lexigram/web/quickstart/core.py: 29%
103 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Quick-start module for single-file Lexigram web applications.
3Provides a module-level ``app`` ASGI instance so that a minimal API can be
4built in 6 lines::
6 from lexigram.web import app, get, singleton
8 @singleton
9 class UserRepo:
10 async def find(self, id: str) -> dict:
11 return {"id": id, "name": "Alice"}
13 @get("/users/{id}")
14 async def get_user(id: str, repo: UserRepo) -> dict:
15 return await repo.find(id)
17Routes registered with ``@get``, ``@post``, etc. are automatically picked up
18by the :class:`~lexigram.web.routing.route_handlers.CoreRouteHandler` when the
19application boots.
21The ``app`` object is a lazy ASGI proxy — the real application is only created
22when the first ASGI message arrives.
24:func:`singleton` and :func:`injectable` decorators register classes directly
25in a module-level registry so the quickstart container can discover them even
26when they are defined inside functions or test scopes (not as module-level
27names found by a ``sys.modules`` scan).
28"""
30from __future__ import annotations
32from dataclasses import dataclass
33from typing import TYPE_CHECKING, Any, TypeVar
35from lexigram.logging import get_logger
37if TYPE_CHECKING:
38 from collections.abc import Callable
40T = TypeVar("T")
42logger = get_logger(__name__)
44# ---------------------------------------------------------------------------
45# Quickstart DI registry
46# ---------------------------------------------------------------------------
48#: Global registry populated by :func:`singleton` and :func:`injectable`.
49#: Entries are ``(cls, scope_string)`` pairs consumed at boot time by
50#: :class:`_QuickstartApp._collect_script_services`.
51_QUICKSTART_REGISTRY: list[tuple[type, str]] = []
54def singleton(cls: type[T]) -> type[T]:
55 """Mark *cls* as a singleton and auto-register it in the quickstart container.
57 Combines the ``lexigram.di.singleton`` DI marker (which sets
58 ``__lexigram_injectable__``) with a direct entry in the quickstart
59 service registry so the class is discovered regardless of whether it
60 is defined at module scope.
62 Returns the class unchanged, so the decorator is transparent.
64 Args:
65 cls: The class to register as a container-managed singleton.
67 Returns:
68 The original *cls* with the injectable marker applied.
70 Example::
72 from lexigram.web import app, get, singleton
74 @singleton
75 class UserRepo:
76 async def find(self, user_id: str) -> dict:
77 return {"id": user_id}
78 """
79 from lexigram.contracts.core.scopes import ServiceScope
80 from lexigram.di.decorators import Injectable
82 Injectable(scope=ServiceScope.SINGLETON)(cls)
83 _QUICKSTART_REGISTRY.append((cls, ServiceScope.SINGLETON))
84 return cls
87def injectable(cls: type[T]) -> type[T]:
88 """Mark *cls* as transient and auto-register it in the quickstart container.
90 Each DI resolution creates a new instance of the class (transient scope).
91 This is the quickstart alias for the transient-scope decorator.
93 Returns the class unchanged, so the decorator is transparent.
95 Args:
96 cls: The class to register as a transient (per-resolve) service.
98 Returns:
99 The original *cls* with the injectable marker applied.
101 Example::
103 from lexigram.web import app, get, injectable
104 from lexigram.logging import get_logger
106 logger = get_logger(__name__)
108 @injectable
109 class RequestLogger:
110 def log(self, msg: str) -> None:
111 logger.info("request_log", message=msg)
112 """
113 from lexigram.contracts.core.scopes import ServiceScope
114 from lexigram.di.decorators import Injectable
116 Injectable(scope=ServiceScope.TRANSIENT)(cls)
117 _QUICKSTART_REGISTRY.append((cls, ServiceScope.TRANSIENT))
118 return cls
121def _reset_quickstart_registry() -> None:
122 """Clear the quickstart service registry and reset the global application state.
124 Intended for use in tests to prevent cross-test state pollution.
125 """
126 _QUICKSTART_REGISTRY.clear()
128 # Also reset the global QuickstartApp state if it exists
129 if "app" in globals():
130 app_obj = globals()["app"]
131 if isinstance(app_obj, _QuickstartApp):
132 app_obj._booted = False
133 app_obj._application = None
134 app_obj._starlette = None
137# ---------------------------------------------------------------------------
138# Internal data types
139# ---------------------------------------------------------------------------
142@dataclass
143class _PendingRoute:
144 """Lightweight route definition for script-mode handlers."""
146 path: str
147 method: str
148 handler: Callable[..., Any]
151#: Global registry for routes in quickstart mode.
152#: Previously populated implicitly via sys.modules scan. Now must be set
153#: explicitly on :class:`_QuickstartApp` instance before boot.
154_QUICKSTART_ROUTE_REGISTRY: list[_PendingRoute] = []
157class _QuickstartApp:
158 """Lazy ASGI proxy for **development and prototyping only**.
160 On first ASGI call the real :class:`~lexigram.app.base.Application` and
161 :class:`~lexigram.web.di.provider.WebProvider` are created and booted.
162 All subsequent calls are forwarded directly to the underlying Starlette app.
164 .. warning::
165 Quickstart mode is intended for **single-file scripts and prototypes**.
166 It maintains global module-level state (route registry, service registry)
167 that does not support multi-module projects, hot-reload isolation, or
168 running multiple applications in the same process.
170 For production applications use the full provider pattern::
172 from lexigram.app import Application
173 from lexigram.web.di.provider import WebProvider
175 app = Application()
176 app.add_provider(WebProvider(controllers=[...]))
177 """
179 def __init__(self) -> None:
180 import os
182 env = os.environ.get("LEX_ENV", os.environ.get("APP_ENV", "development"))
183 if env == "production":
184 import warnings
186 warnings.warn(
187 "lexigram.web quickstart module is loaded in a production environment "
188 "(LEX_ENV=production). Quickstart is for development only. "
189 "Use the full Application + WebProvider pattern in production.",
190 stacklevel=2,
191 category=UserWarning,
192 )
193 self._starlette: Any = None
194 self._application: Any = None
195 self._booted: bool = False
197 def _collect_script_routes(self) -> list[_PendingRoute]:
198 """Collect routes from explicit registry and sys.modules (backward compatibility).
200 Returns explicit routes first, then falls back to sys.modules scan for
201 backward compatibility with code that relies on automatic route discovery.
203 Returns:
204 A list of :class:`_PendingRoute` instances.
205 """
206 import sys
208 routes: list[_PendingRoute] = list(_QUICKSTART_ROUTE_REGISTRY)
209 seen: set[int] = {id(r.handler) for r in routes}
211 for mod in list(sys.modules.values()):
212 module_name: str = getattr(mod, "__name__", "") or ""
213 # Skip framework internals to avoid double-registration
214 if module_name.startswith("lexigram"):
215 continue
217 try:
218 module_vars = vars(mod)
219 except TypeError:
220 continue
222 for attr in module_vars.values():
223 if not callable(attr):
224 continue
226 fn_id = id(attr)
227 if fn_id in seen:
228 continue
230 # Case 1: Function-based routes (@get, @post, etc)
231 # Strict check to avoid Mock objects (REVISION SCR-4)
232 config = getattr(attr, "_route_config", None)
233 if isinstance(config, dict):
234 seen.add(fn_id)
235 routes.append(
236 _PendingRoute(
237 path=config["path"],
238 method=config["method"],
239 handler=attr,
240 )
241 )
242 continue
244 # Case 2: Class-based WebSocket handlers (@websocket_handler)
245 # Strict check to avoid Mock objects (REVISION SCR-5)
246 is_ws = getattr(attr, "_is_websocket_handler", False)
247 if is_ws is True:
248 path = getattr(attr, "_ws_path", None)
249 if isinstance(path, str):
250 seen.add(fn_id)
251 routes.append(
252 _PendingRoute(
253 path=path,
254 method="WEBSOCKET",
255 handler=attr,
256 )
257 )
258 continue
259 return routes
261 def _collect_script_services(self) -> list[tuple[type, Any]]:
262 """Return services from the explicit registry.
264 Previously this method combined the explicit quickstart registry with
265 a sys.modules scan for module-level @injectable/@singleton classes.
266 Now it returns only services from the explicit registry, making service
267 discovery deterministic and not subject to module import order.
269 Returns:
270 A list of ``(cls, scope_string)`` pairs from the quickstart registry.
271 """
272 return list(_QUICKSTART_REGISTRY)
274 async def _ensure_booted(self) -> None:
275 if self._booted:
276 return
277 from lexigram.app.base import Application
278 from lexigram.identity.di.provider import IdentityProvider
279 from lexigram.observability.di.sub_providers.observability import (
280 ObservabilityProvider,
281 )
282 from lexigram.web.config import WebConfig
283 from lexigram.web.di.provider import WebProvider
284 from lexigram.web.security.config import CSRFConfig
286 # Script mode: disable CSRF (pure API, no browser session auth)
287 web_config = WebConfig()
288 web_config.security.csrf = CSRFConfig(enabled=False)
290 # Add core infrastructure providers first (web depends on them)
291 self._application = Application(name="lexigram-quickstart")
292 self._application.add_provider(IdentityProvider())
293 self._application.add_provider(ObservabilityProvider())
295 provider = WebProvider(web_config=web_config)
296 # Pass @singleton / @injectable services collected from user modules
297 provider._extra_injectable_services = self._collect_script_services()
298 # Inject script-mode routes so CoreRouteHandler can discover them
299 self._application._pending_routes = self._collect_script_routes()
300 self._application.add_provider(provider)
301 await self._application.start()
302 self._starlette = provider.starlette
303 logger.warning(
304 "quickstart_cold_boot",
305 message="First request triggered application boot. "
306 "Use create_app() for production deployments.",
307 )
308 self._booted = True
310 async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
311 await self._ensure_booted()
312 await self._starlette(scope, receive, send)
315#: Module-level ASGI-compatible application.
316#: Import as ``from lexigram.web import app``.
317app: _QuickstartApp = _QuickstartApp()
320__all__ = [
321 "_PendingRoute",
322 "_QuickstartApp",
323 "_reset_quickstart_registry",
324 "app",
325 "injectable",
326 "singleton",
327]