Coverage for src / lexigram / ai / relay / gateway / web / audio_endpoints.py: 94%
121 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""Audio passthrough endpoints for the relay gateway web layer.
3The three OpenAI-shaped audio endpoints — speech synthesis, audio
4transcription, and audio translation — are served through
5``PassthroughService`` exactly like the generic passthrough routes, with
6two differences: transcription/translation submissions arrive as
7``multipart/form-data`` (raw bytes with the model lifted from the
8``model`` form field and the body forwarded byte-for-byte) and upstream
9audio responses are returned as their raw body bytes verbatim instead of
10being dropped. JSON submissions (speech) keep the decoded-object path
11with ``RelayPassthroughBody.json``; multipart submissions carry their
12original content type so the provider sees the boundary untouched.
13Failures render in the OpenAI error envelope through the same machinery
14as the other passthrough routes.
16This module owns its route table and endpoint handlers only; mounting
17them to the web contributor is handled by the routes coordinator.
18"""
20from __future__ import annotations
22from collections.abc import Awaitable, Callable, Mapping
23from typing import Any, TypeAlias
25from starlette.requests import Request
26from starlette.responses import JSONResponse, Response
28from lexigram.ai.relay.gateway.passthrough import (
29 PassthroughService,
30 RelayPassthroughBody,
31 RelayPassthroughResult,
32 rewrite_multipart_form_field,
33)
34from lexigram.contracts.ai.relay import (
35 RelayFormat,
36 RelayGatewayError,
37 RelayGatewayRequest,
38)
39from lexigram.contracts.ai.relay.gateway import RelayGatewayErrorCode
40from lexigram.identity.ambient import new_uuid
41from lexigram.serialization import loads
43__all__ = [
44 "AUDIO_ROUTE_TABLE",
45 "audio_speech_endpoint",
46 "audio_transcriptions_endpoint",
47 "audio_translations_endpoint",
48]
50ResolvePassthrough: TypeAlias = Callable[[Request], Awaitable[PassthroughService]]
52AUDIO_ROUTE_TABLE: tuple[tuple[str, str], ...] = (
53 ("/v1/audio/speech", "audio_speech"),
54 ("/v1/audio/transcriptions", "audio_transcriptions"),
55 ("/v1/audio/translations", "audio_translations"),
56)
57"""Inbound path to endpoint kind for the audio passthrough routes."""
59_MULTIPART_MEDIA_TYPE = "multipart/form-data"
61_HOP_BY_HOP_HEADERS: frozenset[str] = frozenset(
62 {
63 "connection",
64 "keep-alive",
65 "proxy-authenticate",
66 "proxy-authorization",
67 "te",
68 "trailer",
69 "transfer-encoding",
70 "upgrade",
71 }
72)
73"""Hop-by-hop headers that must never be relayed to clients."""
75_ERROR_TYPE_MAP: dict[int, str] = {
76 400: "invalid_request_error",
77 401: "authentication_error",
78 403: "permission_denied_error",
79 404: "invalid_request_error",
80 409: "conflict_error",
81 429: "rate_limit_error",
82 499: "cancelled_error",
83 502: "server_error",
84 504: "server_error",
85}
86"""Per-status OpenAI error type names for the audio error envelope."""
88_FORM_FIELD_HEADER_MARKER = b'name="'
89_FORM_FIELD_HEADER_SUFFIX = b'"'
90"""Multipart ``Content-Disposition`` attribute delimiters used by field lookups."""
93def _content_type(request: Request) -> str:
94 """Return the request's content-type header value or empty string."""
95 return request.headers.get("content-type", "")
98def _is_multipart(content_type: str) -> bool:
99 """Tell whether a content-type value denotes ``multipart/form-data``.
101 Args:
102 content_type: A content-type header value.
104 Returns:
105 ``True`` for the ``multipart/form-data`` media type regardless
106 of parameters (e.g. the boundary).
107 """
108 return content_type.partition(";")[0].strip().lower() == _MULTIPART_MEDIA_TYPE
111def _multipart_boundary(content_type: str) -> str | None:
112 """Extract the ``boundary`` parameter from a content-type header.
114 Args:
115 content_type: The raw content-type header value.
117 Returns:
118 The boundary token without surrounding quotes, or ``None`` when
119 the multipart header carries no boundary parameter.
120 """
121 for parameter in content_type.split(";"):
122 key, separator, raw_value = parameter.strip().partition("=")
123 if separator and key.lower() == "boundary" and raw_value.strip('"'):
124 return raw_value.strip().strip('"')
125 return None
128def _extract_multipart_model(body: bytes, boundary: str) -> str | None:
129 """Extract the ``model`` form field value from a multipart body.
131 Narrow boundary-aware lookup mirroring the passthrough rewrite
132 helper: the body is split on the ``--<boundary>`` framing marker and
133 the first part whose ``Content-Disposition`` header carries
134 ``name="model"`` has its value content read.
136 Args:
137 body: The raw ``multipart/form-data`` body bytes.
138 boundary: The boundary token from the content-type header.
140 Returns:
141 The model field value, or ``None`` when the body has no
142 ``model`` field or no boundary marker.
143 """
144 marker = b"--" + boundary.encode("utf-8")
145 target = _FORM_FIELD_HEADER_MARKER + b"model" + _FORM_FIELD_HEADER_SUFFIX
146 segments = body.split(marker)
147 if len(segments) < 2:
148 return None
149 for index in range(1, len(segments) - 1):
150 part = segments[index]
151 separator = part.find(b"\r\n\r\n")
152 if separator < 0:
153 continue
154 headers = part[2:separator].lower()
155 if target not in headers:
156 continue
157 value_end = len(part) - 2 if part.endswith(b"\r\n") else len(part)
158 return part[separator + 4 : value_end].decode("utf-8", "ignore").strip()
159 return None
162def _parse_json(raw: bytes) -> dict[str, Any] | None:
163 """Decode a JSON request body into an object, or ``None`` when malformed.
165 Args:
166 raw: The raw request body bytes.
168 Returns:
169 The decoded JSON object, or ``None`` for malformed JSON and
170 non-object roots.
171 """
172 try:
173 decoded = loads(raw)
174 except (TypeError, ValueError):
175 return None
176 if not isinstance(decoded, dict):
177 return None
178 return decoded
181def _tenant_id(request: Request) -> str:
182 """Read the tenant id from the auth middleware's normalized user dict.
184 Args:
185 request: The Starlette request being served.
187 Returns:
188 The ``tenant_id`` (or ``tenant``) value when the state user is a
189 dict carrying one, otherwise an empty string.
190 """
191 user = getattr(request.state, "user", None)
192 if not isinstance(user, dict):
193 return ""
194 tenant = user.get("tenant_id") or user.get("tenant")
195 if isinstance(tenant, str):
196 return tenant
197 return ""
200def _error_response(request_id: str, error: RelayGatewayError) -> Response:
201 """Build the OpenAI error envelope for a gateway failure.
203 Never includes request payloads, headers, or tracebacks; only the
204 safe error fields the protocol documents.
206 Args:
207 request_id: Request id stamped on the error.
208 error: The gateway error to render.
210 Returns:
211 A JSON response with the OpenAI error envelope and the error's
212 status code.
213 """
214 return JSONResponse(
215 content={
216 "error": {
217 "message": error.message,
218 "type": _ERROR_TYPE_MAP.get(error.status_code, "server_error"),
219 "code": error.code,
220 "request_id": request_id,
221 }
222 },
223 status_code=error.status_code,
224 )
227def _invalid_request(request_id: str, message: str) -> Response:
228 """Build a 400 OpenAI error envelope for a malformed request.
230 Args:
231 request_id: Request id stamped on the error.
232 message: Safe error message.
234 Returns:
235 A 400 JSON response with the OpenAI error envelope.
236 """
237 return _error_response(
238 request_id,
239 RelayGatewayError(
240 code=RelayGatewayErrorCode.INVALID_REQUEST,
241 message=message,
242 status_code=400,
243 request_id=request_id,
244 ),
245 )
248def _safe_headers(
249 headers: Mapping[str, str], request_id: str, trace_id: str
250) -> dict[str, str]:
251 """Filter result headers and stamp request metadata.
253 Drops ``set-cookie`` and all hop-by-hop headers case-insensitively,
254 keeps everything else, and always adds ``x-request-id`` plus
255 ``x-trace-id`` when a trace id was provided.
257 Args:
258 headers: The result headers to filter.
259 request_id: Request id stamped as ``x-request-id``.
260 trace_id: Trace id stamped as ``x-trace-id`` when non-empty.
262 Returns:
263 The safe header dict.
264 """
265 safe: dict[str, str] = {}
266 for key, value in headers.items():
267 lowered = key.lower()
268 if lowered == "set-cookie" or lowered in _HOP_BY_HOP_HEADERS:
269 continue
270 safe[key] = value
271 safe["x-request-id"] = request_id
272 if trace_id:
273 safe["x-trace-id"] = trace_id
274 return safe
277def _render_result(
278 ok_result: RelayPassthroughResult,
279 request_id: str,
280 trace_id: str,
281) -> Response:
282 """Render a passthrough result as JSON or verbatim raw bytes.
284 JSON responses keep their decoded payload through the ``payload``
285 accessor; non-JSON responses (e.g. ``audio/mpeg``) ride in ``body``
286 uninterpreted with their content type.
288 Args:
289 ok_result: The successful passthrough result.
290 request_id: Request id stamped as ``x-request-id``.
291 trace_id: Trace id stamped as ``x-trace-id`` when non-empty.
293 Returns:
294 ``JSONResponse`` for JSON results, a raw ``Response`` carrying
295 the body bytes for audio results, or ``204`` when the result has
296 neither payload nor body.
297 """
298 headers = _safe_headers(ok_result.headers, request_id, trace_id)
299 if ok_result.payload is not None:
300 return JSONResponse(
301 content=ok_result.payload,
302 status_code=ok_result.status_code,
303 headers=headers,
304 )
305 if ok_result.body:
306 headers.setdefault(
307 "content-type", ok_result.content_type or "application/octet-stream"
308 )
309 return Response(
310 content=ok_result.body,
311 status_code=ok_result.status_code,
312 headers=headers,
313 )
314 return Response(status_code=204, headers=headers)
317async def _handle_audio(
318 kind: str,
319 resolve_passthrough: ResolvePassthrough,
320 request: Request,
321) -> Response:
322 """Serve one audio passthrough request.
324 The request body is read exactly once and shaped into a
325 ``RelayPassthroughBody``: decoded JSON for JSON submissions, or raw
326 bytes with the content type intact for ``multipart/form-data``
327 submissions (with the ``model`` field rewritten from the extracted
328 value). The passthrough service is resolved per request, never
329 cached.
331 Args:
332 kind: The endpoint kind owned by this route.
333 resolve_passthrough: Resolver of the passthrough service.
334 request: The Starlette request being served.
336 Returns:
337 The upstream payload rendered (JSON or raw audio bytes), or the
338 OpenAI error envelope for gateway failures and malformed
339 requests.
340 """
341 raw = await request.body()
342 request_id = getattr(request.state, "request_id", None) or new_uuid()
343 trace_id = request.headers.get("x-trace-id", "") or ""
344 content_type = _content_type(request)
345 if _is_multipart(content_type):
346 boundary = _multipart_boundary(content_type)
347 if boundary is None:
348 return _invalid_request(
349 request_id, "multipart body must declare a boundary"
350 )
351 model = _extract_multipart_model(raw, boundary)
352 if model is None or not model:
353 return _invalid_request(request_id, "model is required")
354 payload = RelayPassthroughBody.raw(
355 rewrite_multipart_form_field(raw, boundary, "model", model),
356 content_type,
357 )
358 else:
359 decoded = _parse_json(raw)
360 if decoded is None:
361 return _invalid_request(request_id, "malformed JSON body")
362 model = decoded.get("model")
363 if not isinstance(model, str) or not model:
364 return _invalid_request(request_id, "model is required")
365 payload = RelayPassthroughBody.json(decoded)
366 gateway_request = RelayGatewayRequest(
367 request_id=request_id,
368 tenant_id=_tenant_id(request),
369 source=RelayFormat.OPENAI_CHAT,
370 model=model,
371 stream=False,
372 payload=payload,
373 headers=dict(request.headers.items()),
374 channel=None,
375 )
376 service = await resolve_passthrough(request)
377 result = await service.handle(kind, gateway_request)
378 if result.is_err():
379 return _error_response(request_id, result.unwrap_err())
380 ok_result = result.unwrap()
381 return _render_result(ok_result, request_id, trace_id)
384async def audio_speech_endpoint(
385 resolve_passthrough: ResolvePassthrough,
386 request: Request,
387) -> Response:
388 """Serve ``POST /v1/audio/speech``.
390 JSON submissions (``{model, input, voice}``) forward decoded with
391 the model resolved from the body; the upstream ``audio/mpeg``
392 response returns verbatim.
394 Args:
395 resolve_passthrough: Resolver of the passthrough service.
396 request: The Starlette request being served.
398 Returns:
399 The upstream payload rendered (JSON or raw audio), or the OpenAI
400 error envelope for failures.
401 """
402 return await _handle_audio("audio_speech", resolve_passthrough, request)
405async def audio_transcriptions_endpoint(
406 resolve_passthrough: ResolvePassthrough,
407 request: Request,
408) -> Response:
409 """Serve ``POST /v1/audio/transcriptions``.
411 Multipart submissions forward raw bytes with a boundary-aware
412 ``model`` field rewrite; JSON submissions follow the decoded path.
414 Args:
415 resolve_passthrough: Resolver of the passthrough service.
416 request: The Starlette request being served.
418 Returns:
419 The upstream payload rendered as JSON or raw audio bytes, or the
420 OpenAI error envelope for failures.
421 """
422 return await _handle_audio("audio_transcriptions", resolve_passthrough, request)
425async def audio_translations_endpoint(
426 resolve_passthrough: ResolvePassthrough,
427 request: Request,
428) -> Response:
429 """Serve ``POST /v1/audio/translations``.
431 Multipart submissions forward raw with the model lifted from the
432 ``model`` form field; JSON submissions follow the decoded path.
434 Args:
435 resolve_passthrough: Resolver of the passthrough service.
436 request: The Starlette request being served.
438 Returns:
439 The upstream payload rendered as JSON or raw audio bytes, or the
440 OpenAI error envelope for failures.
441 """
442 return await _handle_audio("audio_translations", resolve_passthrough, request)