Coverage for src / lexigram / ai / relay / gateway / web / shared.py: 96%
49 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""Shared request/response helpers for the relay gateway web layer.
3Error envelopes, body parsing, and header filtering are common to every
4inbound route family (chat relay, passthrough, audio, images, job
5relay). They live here so endpoint modules can share them without
6importing from ``routes`` and forming an import cycle.
7"""
9from __future__ import annotations
11from collections.abc import Awaitable, Callable, Mapping
12from typing import Any, TypeAlias
14from starlette.requests import Request
15from starlette.responses import JSONResponse, Response
17from lexigram.ai.relay.gateway.passthrough import PassthroughService
18from lexigram.contracts.ai.relay import RelayFormat, RelayGatewayError
19from lexigram.contracts.ai.relay.gateway import RelayGatewayErrorCode
20from lexigram.serialization import loads
22__all__ = [
23 "_DEFAULT_ERROR_TYPES",
24 "_ERROR_TYPE_MAP",
25 "_HOP_BY_HOP_HEADERS",
26 "ResolvePassthrough",
27 "_error_response",
28 "_error_types",
29 "_parse_body",
30 "_safe_headers",
31]
33ResolvePassthrough: TypeAlias = Callable[[Request], Awaitable[PassthroughService]]
34"""Resolver of a passthrough service from a Starlette request."""
36_HOP_BY_HOP_HEADERS: frozenset[str] = frozenset(
37 {
38 "connection",
39 "keep-alive",
40 "proxy-authenticate",
41 "proxy-authorization",
42 "te",
43 "trailer",
44 "transfer-encoding",
45 "upgrade",
46 }
47)
48"""Hop-by-hop headers that must never be relayed to clients."""
50_ERROR_TYPE_MAP: dict[int, tuple[str, str, str]] = {
51 400: ("invalid_request_error", "invalid_request_error", "INVALID_ARGUMENT"),
52 401: ("authentication_error", "authentication_error", "UNAUTHENTICATED"),
53 403: ("permission_denied_error", "permission_denied_error", "PERMISSION_DENIED"),
54 404: ("invalid_request_error", "not_found_error", "NOT_FOUND"),
55 409: ("conflict_error", "conflict_error", "FAILED_PRECONDITION"),
56 429: ("rate_limit_error", "rate_limit_error", "RESOURCE_EXHAUSTED"),
57 499: ("cancelled_error", "cancelled_error", "CANCELLED"),
58 502: ("server_error", "api_error", "INTERNAL"),
59 504: ("server_error", "api_error", "DEADLINE_EXCEEDED"),
60}
61"""Per-status error type names for the OpenAI, Claude, and Google families."""
63_DEFAULT_ERROR_TYPES: tuple[str, str, str] = ("server_error", "api_error", "INTERNAL")
64"""Error type names for every unmapped status code (including 500)."""
67def _error_types(status_code: int) -> tuple[str, str, str]:
68 """Map an HTTP status to per-protocol error type names.
70 Args:
71 status_code: The gateway error's status code.
73 Returns:
74 ``(openai_type, claude_type, google_status)`` for the status;
75 the server-error triple for unmapped statuses.
76 """
77 return _ERROR_TYPE_MAP.get(status_code, _DEFAULT_ERROR_TYPES)
80def _error_response(source: RelayFormat, error: RelayGatewayError) -> Response:
81 """Build the inbound-protocol error envelope for a gateway error.
83 Never includes request payloads, headers, or tracebacks; only the
84 safe error fields the protocol documents.
86 Args:
87 source: The inbound wire format determining the envelope shape.
88 error: The gateway error to render.
90 Returns:
91 A JSON response with the protocol's error envelope and the
92 error's status code.
93 """
94 openai_type, claude_type, google_status = _error_types(error.status_code)
95 if source == RelayFormat.OPENAI_CHAT:
96 envelope: dict[str, Any] = {
97 "error": {
98 "message": error.message,
99 "type": openai_type,
100 "code": error.code,
101 "request_id": error.request_id,
102 }
103 }
104 elif source == RelayFormat.OPENAI_RESPONSES:
105 envelope = {
106 "error": {
107 "message": error.message,
108 "type": openai_type,
109 "code": error.code,
110 }
111 }
112 elif source == RelayFormat.CLAUDE:
113 envelope = {
114 "type": "error",
115 "error": {"type": claude_type, "message": error.message},
116 }
117 else:
118 envelope = {
119 "error": {
120 "code": error.status_code,
121 "message": error.message,
122 "status": google_status,
123 }
124 }
125 return JSONResponse(content=envelope, status_code=error.status_code)
128def _parse_body(
129 raw: bytes, source: RelayFormat, request_id: str
130) -> dict[str, Any] | Response:
131 """Decode the request body, returning a 400 response when malformed.
133 Args:
134 raw: The raw request body bytes.
135 source: The inbound wire format for the error envelope.
136 request_id: Request id stamped on the error.
138 Returns:
139 The decoded JSON object, or a 400 ``INVALID_REQUEST`` response
140 for malformed JSON, non-object roots, and empty bodies.
141 """
142 try:
143 decoded = loads(raw)
144 except (TypeError, ValueError):
145 return _error_response(
146 source,
147 RelayGatewayError(
148 code=RelayGatewayErrorCode.INVALID_REQUEST,
149 message="malformed JSON body",
150 status_code=400,
151 request_id=request_id,
152 ),
153 )
154 if not isinstance(decoded, dict):
155 return _error_response(
156 source,
157 RelayGatewayError(
158 code=RelayGatewayErrorCode.INVALID_REQUEST,
159 message="request body must be a JSON object",
160 status_code=400,
161 request_id=request_id,
162 ),
163 )
164 return decoded
167def _safe_headers(
168 headers: Mapping[str, str], request_id: str, trace_id: str
169) -> dict[str, str]:
170 """Filter result headers and stamp request metadata.
172 Drops ``set-cookie`` and all hop-by-hop headers case-insensitively,
173 keeps everything else, and always adds ``x-request-id`` plus
174 ``x-trace-id`` when a trace id was provided.
176 Args:
177 headers: The result headers to filter.
178 request_id: Request id stamped as ``x-request-id``.
179 trace_id: Trace id stamped as ``x-trace-id`` when non-empty.
181 Returns:
182 The safe header dict.
183 """
184 safe: dict[str, str] = {}
185 for key, value in headers.items():
186 lowered = key.lower()
187 if lowered == "set-cookie" or lowered in _HOP_BY_HOP_HEADERS:
188 continue
189 safe[key] = value
190 safe["x-request-id"] = request_id
191 if trace_id:
192 safe["x-trace-id"] = trace_id
193 return safe