Coverage for src/lexigram/web/transport/responses.py: 61%
51 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""Response handling"""
3from __future__ import annotations
5from collections.abc import AsyncGenerator, Iterator
6from pathlib import Path
7from typing import Any, cast
9from starlette.responses import (
10 FileResponse as StarletteFileResponse,
11)
12from starlette.responses import (
13 HTMLResponse as StarletteHTMLResponse, # type: ignore[import]
14)
15from starlette.responses import (
16 RedirectResponse as StarletteRedirectResponse,
17)
18from starlette.responses import (
19 Response as StarletteResponse,
20)
21from starlette.responses import (
22 StreamingResponse as StarletteStreamingResponse,
23)
25from lexigram.logging import get_logger
27# JSON helper for HTMX header encoding
28from lexigram.serialization import dumps, dumps_str
30# Use lexigram common JSON utilities for consistent serialization across packages
32logger = get_logger(__name__)
35class Response(StarletteResponse):
36 """Base response class"""
39class FastJSONResponse(Response):
40 """JSON response using lexigram common JSON utilities.
42 Uses orjson when available (5-10x faster than stdlib json) with graceful
43 fallback to stdlib json. Provides consistent JSON serialization across
44 all Lexigram packages.
46 Performance benefits:
47 - 5-10x faster than stdlib json when orjson is available
48 - Native datetime/UUID/dataclass support
49 - Efficient bytes output
50 - Consistent behavior across packages
52 For Pydantic models, prefer model.model_dump_json() which uses
53 Rust-based serialization (similar performance to orjson).
54 """
56 media_type = "application/json"
58 def render(self, content: Any) -> bytes:
59 # Check if content is a Pydantic model
60 if hasattr(content, "model_dump_json"):
61 # Use Pydantic's efficient Rust-based serializer
62 return content.model_dump_json().encode("utf-8")
64 # Use lexigram common JSON utilities for everything else
65 return cast("Any", dumps)(content)
68class JSONResponse(FastJSONResponse):
69 """JSON response with high performance.
71 Uses lexigram common JSON utilities for consistent serialization
72 across all Lexigram packages. Automatically uses Pydantic's Rust serializer
73 for Pydantic models when available.
74 """
77class HTMLResponse(StarletteHTMLResponse):
78 """HTML response"""
81class HTMLContent(str):
82 """Marker type explicitly indicating HTML content.
84 Handlers that return `HTMLContent` are explicitly declaring that the
85 returned string should be treated as HTML. This avoids heuristic checks
86 and makes the behavior explicit for callers.
87 """
89 __slots__ = ()
92class HTMXResponse(StarletteHTMLResponse):
93 """Convenience HTML response with HTMX-specific headers support.
95 Example:
96 return HTMXResponse("<div>ok</div>", hx_trigger={"showToast": "Saved"})
97 This will set the `HX-Trigger` header to the JSON-encoded value.
98 """
100 def __init__(
101 self,
102 content: Any,
103 status_code: int = 200,
104 headers: dict[str, str] | None = None,
105 hx_trigger: Any = None,
106 hx_refresh: bool = False,
107 ) -> None:
108 # Prepare headers dict
109 headers = headers or {}
110 # Set HX-Trigger header if provided
111 if hx_trigger is not None:
112 # Allow passing a str or a mapping which will be JSON-encoded
113 try:
114 if isinstance(hx_trigger, str):
115 headers["HX-Trigger"] = hx_trigger
116 else:
117 headers["HX-Trigger"] = cast("Any", dumps_str)(hx_trigger)
118 except (TypeError, ValueError) as e:
119 logger.debug("Failed to JSON-encode HX-Trigger header: %s", e)
120 headers["HX-Trigger"] = str(hx_trigger)
122 if hx_refresh:
123 headers["HX-Refresh"] = "true"
125 super().__init__(content=content, status_code=status_code, headers=headers)
128class FileResponse(StarletteFileResponse):
129 """File response for serving static files"""
132class StreamingResponse(StarletteStreamingResponse):
133 """Streaming response for large files or server-sent events"""
136class RedirectResponse(StarletteRedirectResponse):
137 """Redirect response"""
140# Convenience functions
141def json_response(
142 content: Any,
143 status_code: int = 200,
144 headers: dict[str, str] | None = None,
145) -> JSONResponse:
146 """Create a JSON response"""
147 return JSONResponse(content=content, status_code=status_code, headers=headers)
150def html_response(
151 content: str,
152 status_code: int = 200,
153 headers: dict[str, str] | None = None,
154) -> HTMLResponse:
155 """Create an HTML response"""
156 return HTMLResponse(content=content, status_code=status_code, headers=headers)
159def file_response(
160 path: str | Path,
161 status_code: int = 200,
162 headers: dict[str, str] | None = None,
163 media_type: str | None = None,
164 filename: str | None = None,
165 content_disposition_type: str = "attachment",
166) -> FileResponse:
167 """Create a file response"""
168 return FileResponse(
169 path=path,
170 status_code=status_code,
171 headers=headers,
172 media_type=media_type,
173 filename=filename,
174 content_disposition_type=content_disposition_type,
175 )
178def streaming_response(
179 content: AsyncGenerator[bytes, None] | Iterator[bytes],
180 status_code: int = 200,
181 headers: dict[str, str] | None = None,
182 media_type: str | None = None,
183) -> StreamingResponse:
184 """Create a streaming response"""
185 return StreamingResponse(
186 content=content,
187 status_code=status_code,
188 headers=headers,
189 media_type=media_type,
190 )
193def redirect_response(
194 url: str,
195 status_code: int = 302,
196 headers: dict[str, str] | None = None,
197) -> RedirectResponse:
198 """Create a redirect response"""
199 return RedirectResponse(url=url, status_code=status_code, headers=headers)
202__all__ = [
203 "FastJSONResponse",
204 "FileResponse",
205 "HTMLContent",
206 "HTMLResponse",
207 "HTMXResponse",
208 "JSONResponse",
209 "RedirectResponse",
210 "Response",
211 "StreamingResponse",
212 "file_response",
213 "html_response",
214 "json_response",
215 "redirect_response",
216 "streaming_response",
217]