src/genro_asgi/response.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """HTTP response: one flat, buffered, TYTX-aware class.16 17 ``Response`` is a single slotted class — no subclass hierarchy (no18 JSON/HTML/Streaming/File variants). It buffers the body in memory and, as an19 ASGI application, emits exactly two messages (``http.response.start`` +20 ``http.response.body``). It can be built with content or created empty and21 configured through ``set_header``/``set_cookie``/``set_result``/``set_error``22 before being sent.23 24 ``set_result`` dispatches by result type: ``dict``/``list`` → JSON bytes via25 ``genro_tytx.json_dumps`` (or TYTX serialization — media type from26 ``media_types.TRANSPORT_MIME`` — when the bound request is in TYTX mode),27 ``Path`` → file bytes, ``bytes`` → as-is, ``str`` → UTF-8 text, ``None`` →28 empty. ``set_error`` maps an exception to a status:29 ``HTTPException`` subtypes carry their own status; ``ValueError``/``TypeError``30 → 400, ``FileNotFoundError`` → 404, ``PermissionError`` → 403, anything else31 → 500 (logged). ``set_cookie`` appends a ``set-cookie`` header.32 33 The ``request`` binding is optional (``request=None``); every request-dependent34 branch (the TYTX path) guards for its absence.35 """36 37 from __future__ import annotations38 39 import logging40 from collections.abc import Mapping41 from pathlib import Path42 from typing import Any, Literal, cast43 from urllib.parse import quote44 45 from genro_tytx import json_dumps, to_tytx46 47 from .exceptions import HTTPException48 from .media_types import TRANSPORT_MIME49 from .types import Receive, Scope, Send50 51 __all__ = ["Response"]52 53 # Header inputs accepted at construction time.54 HeadersInput = Mapping[str, str] | list[tuple[str, str]] | None55 56 57 class Response:58 """Buffered HTTP response, usable directly as an ASGI application.59 60 Example:61 >>> response = Response(content="Hello", media_type="text/plain")62 >>> await response(scope, receive, send)63 64 # Or create empty and configure:65 >>> response = Response()66 >>> response.set_header("X-Custom", "value")67 >>> response.set_result({"data": 123}) # auto-detects JSON68 >>> await response(scope, receive, send)69 """70 71 __slots__ = ("body", "status_code", "_media_type", "_headers", "request")72 73 media_type: str | None = None74 charset: str = "utf-8"75 76 # Non-HTTPException error types mapped to a status code (else 500).77 ERROR_MAP: dict[str, int] = {78 "ValueError": 400,79 "TypeError": 400,80 "FileNotFoundError": 404,81 "PermissionError": 403,82 }83 84 def __init__(85 self,86 content: bytes | str | None = None,87 status_code: int = 200,88 headers: HeadersInput = None,89 media_type: str | None = None,90 request: Any = None,91 ) -> None:92 """Build a response.93 94 Note:95 Status 204 (No Content) and 304 (Not Modified) must not carry a96 body per RFC 7230; providing content with those codes may be97 rejected or truncated by the ASGI server.98 """99 self.request = request100 self.status_code = status_code101 if headers is None:102 self._headers: list[tuple[str, str]] = []103 elif isinstance(headers, list):104 self._headers = list(headers)105 else:106 self._headers = list(headers.items())107 self._media_type = media_type108 self.body = self._encode_content(content)109 110 effective_media_type = self._media_type if self._media_type is not None else self.media_type111 if effective_media_type is not None:112 header_names = {name.lower() for name, _ in self._headers}113 if "content-type" not in header_names:114 content_type = self._get_content_type()115 if content_type:116 self._headers.append(("content-type", content_type))117 self._add_content_length()118 119 def _encode_content(self, content: bytes | str | None) -> bytes:120 """Encode content to bytes: None → b"", bytes → as-is, str → charset."""121 if content is None:122 return b""123 if isinstance(content, bytes):124 return content125 return content.encode(self.charset)126 127 def _get_content_type(self) -> str | None:128 """Content-Type value; appends the charset for text types lacking one."""129 effective = self._media_type if self._media_type is not None else self.media_type130 if effective is None:131 return None132 if effective.startswith("text/") and "charset" not in effective:133 return f"{effective}; charset={self.charset}"134 return effective135 136 def _add_content_length(self) -> None:137 """Append a content-length header if none is present."""138 header_names = {name.lower() for name, _ in self._headers}139 if "content-length" not in header_names:140 self._headers.append(("content-length", str(len(self.body))))141 142 def _build_headers(self) -> list[tuple[bytes, bytes]]:143 """ASGI headers: names lowercased, latin-1 encoded (HTTP standard)."""144 return [145 (name.lower().encode("latin-1"), value.encode("latin-1"))146 for name, value in self._headers147 ]148 149 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:150 """ASGI interface: send exactly ``http.response.start`` + one body."""151 await send(152 {153 "type": "http.response.start",154 "status": self.status_code,155 "headers": self._build_headers(),156 }157 )158 await send({"type": "http.response.body", "body": self.body})159 160 def set_header(self, name: str, value: str) -> None:161 """Append a response header. Usable before ``set_result``."""162 self._headers.append((name, value))163 164 def set_cookie(165 self,166 key: str,167 value: str = "",168 *,169 max_age: int | None = None,170 path: str = "/",171 domain: str | None = None,172 secure: bool = False,173 httponly: bool = False,174 samesite: str | None = "lax",175 ) -> None:176 """Append a ``set-cookie`` header (the value is URL-encoded)."""177 cookie = f"{key}={quote(value, safe='')}"178 if max_age is not None:179 cookie += f"; Max-Age={max_age}"180 if path:181 cookie += f"; Path={path}"182 if domain:183 cookie += f"; Domain={domain}"184 if secure:185 cookie += "; Secure"186 if httponly:187 cookie += "; HttpOnly"188 if samesite:189 cookie += f"; SameSite={samesite.capitalize()}"190 self.set_header("set-cookie", cookie)191 192 def set_result(self, result: Any, metadata: dict[str, Any] | None = None) -> None:193 """Set the body from a handler result, dispatching by type.194 195 ``dict``/``list`` → JSON bytes (``genro_tytx.json_dumps``), or TYTX196 bytes/text when the bound request is in TYTX mode; ``Path`` → file197 bytes; ``bytes`` → as-is;198 ``str`` → UTF-8 text; ``None`` → empty; anything else → its ``str``.199 A ``media_type`` in ``metadata`` overrides the type-based default.200 """201 override = metadata.get("media_type") if metadata else None202 if isinstance(result, (dict, list)):203 if self.request is not None and self.request.tytx_mode:204 transport = cast(205 Literal["json", "xml", "msgpack"], self.request.tytx_transport or "json"206 )207 encoded = to_tytx(result, transport)208 self.body = encoded if isinstance(encoded, bytes) else encoded.encode("utf-8")209 self._media_type = override or TRANSPORT_MIME[transport]210 else:211 self.body = json_dumps(result)212 self._media_type = override or "application/json"213 elif isinstance(result, Path):214 self.body = result.read_bytes()215 self._media_type = override or "application/octet-stream"216 elif isinstance(result, bytes):217 self.body = result218 self._media_type = override or "application/octet-stream"219 elif isinstance(result, str):220 self.body = result.encode(self.charset)221 self._media_type = override or "text/plain"222 elif result is None:223 self.body = b""224 self._media_type = override or "text/plain"225 else:226 self.body = str(result).encode(self.charset)227 self._media_type = override or "text/plain"228 self._update_content_headers()229 230 def _update_content_headers(self) -> None:231 """Rebuild content-type and content-length after the body changes."""232 self._headers = [233 (name, value)234 for name, value in self._headers235 if name.lower() not in ("content-type", "content-length")236 ]237 content_type = self._get_content_type()238 if content_type:239 self._headers.append(("content-type", content_type))240 self._headers.append(("content-length", str(len(self.body))))241 242 def set_error(self, error: Exception) -> None:243 """Set the response as an error, mapping the exception to a status.244 245 ``HTTPException`` subtypes carry their own status; other types are246 looked up in ``ERROR_MAP`` (default 500, logged). The body is the247 ``{"error": <message>}`` document through ``set_result``.248 """249 if isinstance(error, HTTPException):250 self.status_code = error.status251 else:252 self.status_code = self.ERROR_MAP.get(type(error).__name__, 500)253 if self.status_code == 500:254 logging.getLogger(__name__).exception("Handler error: %s", error)255 self.set_result({"error": str(error)})