Coverage for src/lexigram/web/protocols.py: 96%
52 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Web-layer protocols for pipes and interceptors.
3Defines the interfaces used exclusively within lexigram-web:
5- Pipe layer: ``PipeProtocol``, ``ParamMetadata``, ``body`` helper.
6- Interceptor layer: ``ExecutionContextProtocol`` (sourced from contracts so
7 that ``GuardProtocol`` in contracts can also reference it without a
8 dependency inversion), ``CallHandlerProtocol``, ``WebInterceptorProtocol``,
9 ``WebInterceptorBase``.
10"""
12from __future__ import annotations
14from dataclasses import dataclass
15from typing import Any, Protocol, runtime_checkable
17from lexigram.contracts.web.execution_context import ExecutionContextProtocol
19# =============================================================================
20# Pipe Protocols
21# =============================================================================
24@dataclass
25class ParamMetadata:
26 """Metadata about a parameter being piped.
28 Attributes:
29 name: The parameter name.
30 param_type: The type of parameter (path, query, body, header, cookie, file).
31 expected_type: The expected Python type for the parameter.
32 default: Default value if the parameter is not provided.
33 alias: Optional alias for the parameter.
34 """
36 name: str
37 param_type: str # "path", "query", "body", "header", "cookie", "file"
38 expected_type: type | None = None
39 default: Any = None
40 alias: str | None = None
43@runtime_checkable
44class PipeProtocol(Protocol):
45 """Transforms or validates a single handler parameter.
47 Pipes operate on individual parameters before they reach the handler.
48 They can transform (e.g., parse string to int) or validate
49 (e.g., check that a value is within range).
51 Example:
52 ```python
53 class ParseIntPipe:
54 async def transform(self, value: Any, metadata: ParamMetadata) -> int:
55 if value is None:
56 return metadata.default or 0
57 try:
58 return int(value)
59 except ValueError:
60 raise BadRequestError(f"Invalid integer for {metadata.name}")
61 ```
62 """
64 async def transform(self, value: Any, metadata: ParamMetadata) -> Any:
65 """Transform or validate the value.
67 Args:
68 value: The value to transform/validate.
69 metadata: Metadata about the parameter.
71 Returns:
72 The transformed value.
74 Raises:
75 Exception: On validation failure (typically HTTP 400 Bad Request).
76 """
77 ...
80def body(
81 name: str | None = None,
82 *,
83 default: Any = ...,
84 alias: str | None = None,
85) -> ParamMetadata:
86 """Create a request-body parameter marker for controller handler methods.
88 Returns a ``ParamMetadata`` instance that the web framework recognises as
89 a body-binding annotation. Use it as a default-value sentinel::
91 @post("/users")
92 async def create_user(self, payload: CreateUserRequest = body()) -> Any:
93 ...
95 Args:
96 name: Optional explicit name override for the parameter.
97 default: Default value when the body cannot be parsed.
98 alias: Optional deserialization alias.
100 Returns:
101 A ``ParamMetadata`` configured for ``param_type="body"``.
102 """
103 return ParamMetadata(
104 name=name or "",
105 param_type="body",
106 default=default,
107 alias=alias,
108 )
111# =============================================================================
112# Interceptor Protocols
113# =============================================================================
114@runtime_checkable
115class CallHandlerProtocol(Protocol):
116 """Wraps the next step in the pipeline.
118 Interceptors use this to continue processing to the next interceptor
119 or to the actual handler.
120 """
122 async def handle(self) -> Any:
123 """Continue processing the request.
125 Returns the result of the next handler in the pipeline.
126 """
127 ...
130@runtime_checkable
131class WebInterceptorProtocol(Protocol):
132 """Intercepts request/response flow with full AOP control.
134 Interceptors wrap the entire request→handler→response lifecycle,
135 enabling cross-cutting concerns like logging, caching, response
136 transformation, and timing without modifying handler code.
138 Example:
139 ```python
140 class TimingInterceptor(Interceptor):
141 async def intercept(
142 self, context: ExecutionContextProtocol, next: CallHandlerProtocol
143 ) -> Any:
144 start = time.perf_counter()
145 result = await next.handle()
146 elapsed = time.perf_counter() - start
147 return result
148 ```
149 """
151 async def intercept(
152 self,
153 context: ExecutionContextProtocol,
154 next_handler: CallHandlerProtocol,
155 ) -> Any:
156 """Intercept the request/response flow.
158 Args:
159 context: Provides metadata about the current request.
160 next_handler: Wraps the next step in the pipeline.
162 Returns:
163 The result of the handler (possibly transformed).
165 Notes:
166 - MUST call ``await next_handler.handle()`` to continue the pipeline.
167 - CAN transform the result before returning.
168 - CAN add/modify response headers.
169 - CAN short-circuit by returning early without calling next.
170 """
171 ...
174class WebInterceptorBase(WebInterceptorProtocol):
175 """Base class for interceptors (optional convenience).
177 Provides a no-op implementation that subclasses can override.
178 """
180 async def intercept(
181 self,
182 context: ExecutionContextProtocol,
183 next_handler: CallHandlerProtocol,
184 ) -> Any:
185 """Default implementation just passes through."""
186 return await next_handler.handle()
189# =============================================================================
190# Provider Protocols
191# =============================================================================
194@runtime_checkable
195class WebAppAccessorProtocol(Protocol):
196 """Provides access to the underlying Starlette application.
198 Consumers that only need the Starlette app should depend on this
199 protocol instead of the full WebProvider, decoupling from the
200 provider's full interface.
201 """
203 @property
204 def starlette(self) -> Any:
205 """Return the Starlette application instance.
207 Returns:
208 The Starlette ASGI application, or None if not yet initialized.
209 """
210 ...
213@runtime_checkable
214class ControllerSourceProtocol(Protocol):
215 """Provides the list of registered controller classes.
217 Consumers that only need access to controllers should depend on this
218 protocol instead of the full WebProvider.
219 """
221 @property
222 def controllers(self) -> list[type]:
223 """Return the list of registered controller classes.
225 Returns:
226 A list of controller class types.
227 """
228 ...
231@runtime_checkable
232class ConfigAccessorProtocol(Protocol):
233 """Provides access to web configuration.
235 Consumers that need configuration details should depend on this
236 protocol for minimal coupling to the provider.
237 """
239 @property
240 def web_config(self) -> Any:
241 """Return the web-layer configuration.
243 Returns:
244 WebConfig instance with web-layer settings.
245 """
246 ...
248 @property
249 def provider_config(self) -> Any:
250 """Return the provider-specific configuration.
252 Returns:
253 WebProviderConfig instance with provider settings.
254 """
255 ...
257 @property
258 def debug_routes_auth(self) -> Any:
259 """Return the debug routes authentication handler, if set.
261 Returns:
262 Callable or None.
263 """
264 ...
266 @property
267 def fail_on_route_conflict(self) -> bool:
268 """Whether to raise on duplicate route registration.
270 Returns:
271 True if duplicate routes should raise RuntimeError.
272 """
273 ...
276@runtime_checkable
277class ProviderResourcesProtocol(Protocol):
278 """Provides access to internal provider resources.
280 Used by advanced components like routing and middleware managers
281 that need direct access to router and OpenAPI generator.
282 """
284 @property
285 def router(self) -> Any:
286 """Return the internal Router instance.
288 Returns:
289 Router instance or None if not initialized.
290 """
291 ...
293 @property
294 def openapi_generator(self) -> Any:
295 """Return the OpenAPI generator instance.
297 Returns:
298 OpenAPIGenerator instance or None if not configured.
299 """
300 ...
303@runtime_checkable
304class WebProviderProtocol(
305 WebAppAccessorProtocol,
306 ControllerSourceProtocol,
307 ConfigAccessorProtocol,
308 ProviderResourcesProtocol,
309 Protocol,
310):
311 """Combined protocol for full WebProvider access.
313 This is the complete interface expected by components that need
314 the full capabilities of WebProvider. Consumers that need fewer
315 properties should depend on the more specific protocols above
316 to reduce coupling.
317 """
320__all__ = [
321 # Pipe layer
322 "ParamMetadata",
323 "PipeProtocol",
324 "body",
325 # Interceptor layer
326 "CallHandlerProtocol",
327 "ExecutionContextProtocol",
328 "WebInterceptorBase",
329 "WebInterceptorProtocol",
330 # Provider layer
331 "WebAppAccessorProtocol",
332 "ControllerSourceProtocol",
333 "ConfigAccessorProtocol",
334 "ProviderResourcesProtocol",
335 "WebProviderProtocol",
336]