Coverage for src/lexigram/web/serialization/serializers.py: 26%
137 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Response serializers for different media types."""
3from __future__ import annotations
5from abc import ABC, abstractmethod
6from typing import TYPE_CHECKING, Any
8from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse, Response
10if TYPE_CHECKING:
11 from starlette.requests import Request as StarletteRequest
13 from lexigram.contracts.events import CommandBusProtocol, QueryBusProtocol
14 from lexigram.contracts.mapping import ObjectMapperProtocol
17class AbstractMediaSerializer(ABC):
18 """Protocol for media-specific response serializers."""
20 @abstractmethod
21 def supports(self, media_type: str) -> bool:
22 """Check if this serializer supports the given media type."""
23 ...
25 @abstractmethod
26 def supported_types(self) -> list[str]:
27 """Return list of supported media types."""
28 ...
30 @abstractmethod
31 async def serialize(
32 self,
33 data: Any,
34 request: StarletteRequest,
35 status_code: int = 200,
36 ) -> Response:
37 """Serialize data into a Response."""
38 ...
41class JSONSerializer(AbstractMediaSerializer):
42 """Serializes responses as JSON using high-performance core utilities."""
44 def supports(self, media_type: str) -> bool:
45 return "application/json" in media_type
47 def supported_types(self) -> list[str]:
48 return ["application/json"]
50 async def serialize(
51 self,
52 data: Any,
53 request: StarletteRequest,
54 status_code: int = 200,
55 ) -> JSONResponse:
56 from lexigram.web.transport.responses import JSONResponse
58 # lexigram.web.transport.responses.JSONResponse handles high-performance
59 # serialization (orjson/Pydantic) via core lexigram.serialization.
60 return JSONResponse(content=data, status_code=status_code) # type: ignore[return-value]
63class HTMLSerializer(AbstractMediaSerializer):
64 """Serializes responses as HTML."""
66 def supports(self, media_type: str) -> bool:
67 return "text/html" in media_type
69 def supported_types(self) -> list[str]:
70 return ["text/html"]
72 async def serialize(
73 self,
74 data: Any,
75 request: StarletteRequest,
76 status_code: int = 200,
77 ) -> HTMLResponse:
78 from lexigram.web.transport.responses import HTMLResponse
80 # Convert to string if not already
81 if not isinstance(data, str):
82 data = str(data)
84 return HTMLResponse(content=data, status_code=status_code) # type: ignore[return-value]
87class PlainTextSerializer(AbstractMediaSerializer):
88 """Serializes responses as plain text."""
90 def supports(self, media_type: str) -> bool:
91 return "text/plain" in media_type
93 def supported_types(self) -> list[str]:
94 return ["text/plain"]
96 async def serialize(
97 self,
98 data: Any,
99 request: StarletteRequest,
100 status_code: int = 200,
101 ) -> PlainTextResponse:
102 # Convert to string if not already
103 if not isinstance(data, str):
104 data = str(data)
106 return PlainTextResponse(content=data, status_code=status_code)
109class XMLSerializer(AbstractMediaSerializer):
110 """Serializes responses as XML."""
112 def supports(self, media_type: str) -> bool:
113 return "application/xml" in media_type or "text/xml" in media_type
115 def supported_types(self) -> list[str]:
116 return ["application/xml", "text/xml"]
118 async def serialize(
119 self,
120 data: Any,
121 request: StarletteRequest,
122 status_code: int = 200,
123 ) -> Response:
124 # Convert to string if not already
125 if not isinstance(data, str):
126 data = str(data)
128 return Response(
129 content=data, status_code=status_code, media_type="application/xml"
130 )
133class ResponseSerializer:
134 """High-level orchestrator for response serialization.
136 Handles:
137 1. Result types (Ok/Err)
138 2. CQRS Messages (Command/Query)
139 3. ObjectMapper transformations
140 4. Media-type specific serialization
142 Dependencies are injected via constructor to avoid service locator pattern.
143 """
145 _registry: dict[str, AbstractMediaSerializer] = {}
146 _default_serializer: AbstractMediaSerializer = JSONSerializer()
148 def __init__(
149 self,
150 command_bus: CommandBusProtocol | None = None,
151 query_bus: QueryBusProtocol | None = None,
152 mapper: ObjectMapperProtocol | None = None,
153 ) -> None:
154 """Initialize the serializer with optional injected dependencies.
156 Args:
157 command_bus: Optional CommandBusProtocol for dispatching commands.
158 query_bus: Optional QueryBusProtocol for executing queries.
159 mapper: Optional ObjectMapperProtocol for transformations.
160 """
161 self.command_bus = command_bus
162 self.query_bus = query_bus
163 self.mapper = mapper
165 @classmethod
166 def _get_registry(cls) -> dict[str, AbstractMediaSerializer]:
167 if not cls._registry:
168 # Initialize registry
169 serializers = [
170 JSONSerializer(),
171 HTMLSerializer(),
172 PlainTextSerializer(),
173 XMLSerializer(),
174 ]
175 for serializer in serializers:
176 for mime in serializer.supported_types():
177 cls._registry[mime.lower()] = serializer
178 return cls._registry
180 @classmethod
181 def negotiate(cls, accept_header: str | None) -> AbstractMediaSerializer:
182 """Negotiate the best serializer based on the Accept header."""
183 if not accept_header:
184 return cls._default_serializer
186 registry = cls._get_registry()
187 for part in accept_header.split(","):
188 mime = part.split(";")[0].strip().lower()
189 if mime in registry:
190 return registry[mime]
191 if mime == "*/*":
192 return cls._default_serializer
194 return cls._default_serializer
196 async def serialize(
197 self,
198 result: Any,
199 request: StarletteRequest,
200 route: Any = None,
201 ) -> Response:
202 """Standardize handler result into a Starlette-compatible Response.
204 Uses injected dependencies instead of service locator pattern.
205 """
206 from lexigram.logging import get_logger
207 from lexigram.result import Result
208 from lexigram.web.transport.responses import HTMLContent, JSONResponse
210 logger = get_logger(__name__)
212 # 0. Handle None → 204 No Content
213 # (Status code from decorator is ignored for None, as 204 is semantic for NO content)
214 if result is None:
215 return Response(status_code=204)
217 # Extract status code from route metadata if available
218 status_code = 200
219 if route and hasattr(route, "metadata") and isinstance(route.metadata, dict):
220 status_code = route.metadata.get("status_code", 200)
221 elif isinstance(route, dict):
222 status_code = route.get("status_code", 200)
224 # 1. Handle explicit Response
225 if isinstance(result, Response):
226 return result
228 # 1.5 Handle bytes directly
229 if isinstance(result, (bytes, bytearray)):
230 return Response(content=result, media_type="application/octet-stream")
232 # 1.6 Handle HTMLContent marker (check before bare str since HTMLContent is str subclass)
233 if isinstance(result, HTMLContent):
234 from lexigram.web.transport.responses import HTMLResponse
236 return HTMLResponse(content=str(result))
238 # 1.7 Handle bare strings as HTML (HTMX-friendly)
239 if isinstance(result, str):
240 return Response(content=result, media_type="text/html")
242 # 2. Handle Result Object
243 if isinstance(result, Result):
244 if result.is_ok():
245 result = result.unwrap()
246 # An Ok that wraps a ready-made Response passes through
247 # untouched (cookie-setting endpoints need this).
248 if isinstance(result, Response):
249 return result
250 # Continue to normal serialization for the inner value
251 else:
252 # Let the DomainExceptionFilter handle the error by raising it
253 # UNLESS it's an HTTPError we want to return directly.
254 from lexigram.web.exceptions import HTTPError
256 error = result.unwrap_err()
257 if isinstance(error, HTTPError):
258 return JSONResponse(
259 content={
260 "error": error.code,
261 "message": error.detail,
262 "details": error.details,
263 },
264 status_code=error.status_code,
265 )
266 from lexigram.web.routing.result_bridge import ResultResponseMapper
268 return ResultResponseMapper.error_to_response(error)
270 # 3. Handle CQRS Messages (Command/Query)
271 from lexigram.contracts.events.messages import Command, Query
273 if isinstance(result, (Command, Query)):
274 try:
275 if isinstance(result, Command):
276 if not self.command_bus:
277 raise ValueError("CommandBusProtocol not injected")
278 result = await self.command_bus.dispatch(result)
279 else:
280 if not self.query_bus:
281 raise ValueError("QueryBusProtocol not injected")
282 result = await self.query_bus.execute(result)
284 # Recursively serialize the bus result
285 return await self.serialize(result, request, route)
286 except Exception as e: # noqa: BLE001 — bus dispatch may raise from user-defined command/query handlers; normalise to InternalServerError
287 from lexigram.web.exceptions import InternalServerError
289 logger.exception("Bus dispatch failed")
290 raise InternalServerError(f"Dispatch error: {e}") from e
292 # 4. Handle ObjectMapper Transformation
293 if (
294 route
295 and hasattr(route, "metadata")
296 and "response_model" in route.metadata
297 and result is not None
298 and self.mapper
299 ):
300 response_model = route.metadata["response_model"]
301 try:
302 if self.mapper:
303 result = self.mapper.map(result, response_model)
304 except (LookupError, RuntimeError, AttributeError):
305 logger.debug("Mapping failed for %s", response_model)
307 # 5.5. Coerce DomainModel → dict for uniform JSON serialization
308 try:
309 from lexigram.contracts.domain.base import ( # type: ignore[attr-defined]
310 DomainModel,
311 )
313 if isinstance(result, DomainModel):
314 result = (
315 result.model_dump()
316 if hasattr(result, "model_dump")
317 else vars(result)
318 )
319 except ImportError:
320 pass
322 # 6. Delegate to media-specific serializers based on Accept header
323 accept = request.headers.get("accept")
324 serializer = self.negotiate(accept)
325 return await serializer.serialize(result, request, status_code=status_code)