Coverage for src / lexigram / ai / relay / gateway / web / routes.py: 95%
163 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-08 23:08 +0800
1"""Inbound relay HTTP routes for the gateway web layer.
3Each route owns one inbound wire format (OpenAI Chat, OpenAI Responses,
4Claude, Gemini) and serves it through a shared endpoint. The gateway
5implementation is resolved at request time from the request-scoped DI
6container. Buffered results return JSON; streaming results return SSE
7frames in the client's own protocol; failures render in the inbound
8protocol's error envelope with safe, filtered headers.
10Passthrough routes (``POST /v1/embeddings``, ``/v1/rerank``,
11``/v1/moderations``, the ``/v1/audio/*`` and ``/v1/images/*`` routes)
12serve non-chat endpoint kinds through ``PassthroughService`` with the
13same request resolution, header filtering, and error envelope machinery,
14without any wire-format conversion. Job-relay routes (``POST
15/v1/videos`` and ``GET /v1/videos/{job_id}``) serve submit-then-poll
16endpoint kinds through ``JobPassthroughService`` with the same envelope
17machinery.
18"""
20from __future__ import annotations
22from collections.abc import AsyncIterator, Awaitable, Callable
23from functools import partial
24from typing import Any, TypeAlias
26from starlette.requests import Request
27from starlette.responses import JSONResponse, Response, StreamingResponse
28from starlette.routing import Route
30from lexigram.ai.relay.gateway.job_passthrough import JobPassthroughService
31from lexigram.ai.relay.gateway.web.audio_endpoints import (
32 AUDIO_ROUTE_TABLE,
33 audio_speech_endpoint,
34 audio_transcriptions_endpoint,
35 audio_translations_endpoint,
36)
37from lexigram.ai.relay.gateway.web.image_endpoints import (
38 IMAGE_ROUTE_TABLE,
39 build_image_routes,
40)
41from lexigram.ai.relay.gateway.web.shared import (
42 ResolvePassthrough,
43 _error_response,
44 _parse_body,
45 _safe_headers,
46)
47from lexigram.ai.relay.gateway.web.sse import SSEEncoder
48from lexigram.contracts.ai.relay import (
49 RelayFormat,
50 RelayGatewayError,
51 RelayGatewayProtocol,
52 RelayGatewayRequest,
53 RelayWireEvent,
54)
55from lexigram.contracts.ai.relay.gateway import RelayGatewayErrorCode
56from lexigram.identity.ambient import new_uuid
58__all__ = ["RELAY_ROUTE_PATHS", "build_routes", "relay_endpoint"]
60ResolveGateway: TypeAlias = Callable[[Request], Awaitable[RelayGatewayProtocol]]
61"""Resolver of a gateway implementation from a Starlette request."""
63ResolveJobPassthrough: TypeAlias = Callable[[Request], Awaitable[JobPassthroughService]]
64"""Resolver of a job passthrough service from a Starlette request."""
66_ROUTE_TABLE: tuple[tuple[str, RelayFormat], ...] = (
67 ("/v1/chat/completions", RelayFormat.OPENAI_CHAT),
68 ("/v1/responses", RelayFormat.OPENAI_RESPONSES),
69 ("/v1/messages", RelayFormat.CLAUDE),
70 ("/v1beta/models/{model}:generateContent", RelayFormat.GEMINI),
71)
72"""Inbound path to wire format ownership per route."""
74_PASSTHROUGH_ROUTE_TABLE: tuple[tuple[str, str], ...] = (
75 ("/v1/embeddings", "embeddings"),
76 ("/v1/rerank", "rerank"),
77 ("/v1/moderations", "moderation"),
78)
79"""Inbound path to endpoint kind for passthrough routes."""
81_JOB_ROUTE_TABLE: tuple[tuple[str, str], ...] = (("/v1/videos", "video_generation"),)
82"""Inbound submit path to endpoint kind for job-relay routes."""
84_JOB_STATUS_PATH = "/v1/videos/{job_id}"
85"""Inbound poll path for the registered job-relay endpoint kinds."""
87_AUDIO_HANDLERS = {
88 "audio_speech": audio_speech_endpoint,
89 "audio_transcriptions": audio_transcriptions_endpoint,
90 "audio_translations": audio_translations_endpoint,
91}
92"""Endpoint kind to handler for the audio passthrough routes."""
94RELAY_ROUTE_PATHS: tuple[str, ...] = tuple(
95 path
96 for path, _ in (
97 *_ROUTE_TABLE,
98 *_PASSTHROUGH_ROUTE_TABLE,
99 *AUDIO_ROUTE_TABLE,
100 *IMAGE_ROUTE_TABLE,
101 )
102)
103"""Inbound relay paths registered by ``build_routes``, in route order."""
106async def relay_endpoint(
107 source: RelayFormat,
108 resolve_gateway: ResolveGateway,
109 request: Request,
110) -> Response:
111 """Serve one inbound relay request in the client's wire protocol.
113 The body is read exactly once, the request id falls back to a
114 generated uuid when the middleware did not set one, and identity
115 comes from the auth middleware's normalized user dict. The gateway
116 is resolved per request, never cached.
118 Args:
119 source: The inbound wire format owned by this route.
120 resolve_gateway: Resolver of the gateway implementation.
121 request: The Starlette request being served.
123 Returns:
124 The protocol-appropriate response: JSON for buffered results,
125 an SSE ``StreamingResponse`` for streaming results, ``204`` for
126 results with neither payload nor stream, or the inbound error
127 envelope for gateway failures.
128 """
129 raw = await request.body()
130 request_id = getattr(request.state, "request_id", None) or new_uuid()
131 trace_id = request.headers.get("x-trace-id", "") or ""
132 body = _parse_body(raw, source, request_id)
133 if isinstance(body, Response):
134 return body
135 stream = bool(body.get("stream", False))
136 model_value: Any = (
137 request.path_params.get("model") if source == RelayFormat.GEMINI else None
138 )
139 if not isinstance(model_value, str) or not model_value:
140 model_value = body.get("model")
141 if not isinstance(model_value, str) or not model_value:
142 return _error_response(
143 source,
144 RelayGatewayError(
145 code=RelayGatewayErrorCode.INVALID_REQUEST,
146 message="model is required",
147 status_code=400,
148 request_id=request_id,
149 ),
150 )
151 user = getattr(request.state, "user", None)
152 tenant_id = ""
153 if isinstance(user, dict):
154 tenant = user.get("tenant_id") or user.get("tenant")
155 if isinstance(tenant, str):
156 tenant_id = tenant
157 gateway_request = RelayGatewayRequest(
158 request_id=request_id,
159 tenant_id=tenant_id,
160 source=source,
161 model=model_value,
162 stream=stream,
163 payload=body,
164 headers=dict(request.headers.items()),
165 channel=None,
166 )
167 gateway = await resolve_gateway(request)
168 result = await gateway.handle(gateway_request)
169 if result.is_err():
170 return _error_response(source, result.unwrap_err())
171 ok_result = result.unwrap()
172 headers = _safe_headers(ok_result.headers, request_id, trace_id)
173 if ok_result.stream is not None:
174 return _streaming_response(source, ok_result.stream, headers)
175 if ok_result.payload is not None:
176 return JSONResponse(
177 content=ok_result.payload,
178 status_code=ok_result.status_code,
179 headers=headers,
180 )
181 return Response(status_code=204, headers=headers)
184async def passthrough_endpoint(
185 kind: str,
186 resolve_passthrough: ResolvePassthrough,
187 request: Request,
188) -> Response:
189 """Serve one inbound passthrough request in its own wire format.
191 The body is read exactly once and forwarded verbatim: no format
192 inference happens, so the request carries the conventional
193 ``OPENAI_CHAT`` source marker and no channel hint. The passthrough
194 service is resolved per request, never cached. Embeddings is an
195 OpenAI-shaped endpoint by convention, so failures render in the
196 OpenAI error envelope through the same machinery as the chat routes.
198 Args:
199 kind: The endpoint kind owned by this route.
200 resolve_passthrough: Resolver of the passthrough service.
201 request: The Starlette request being served.
203 Returns:
204 The upstream JSON verbatim, ``204`` when the result carries
205 neither payload nor stream, or the OpenAI error envelope for
206 gateway failures.
207 """
208 raw = await request.body()
209 request_id = getattr(request.state, "request_id", None) or new_uuid()
210 trace_id = request.headers.get("x-trace-id", "") or ""
211 body = _parse_body(raw, RelayFormat.OPENAI_CHAT, request_id)
212 if isinstance(body, Response):
213 return body
214 model_value = body.get("model")
215 if not isinstance(model_value, str) or not model_value:
216 return _error_response(
217 RelayFormat.OPENAI_CHAT,
218 RelayGatewayError(
219 code=RelayGatewayErrorCode.INVALID_REQUEST,
220 message="model is required",
221 status_code=400,
222 request_id=request_id,
223 ),
224 )
225 user = getattr(request.state, "user", None)
226 tenant_id = ""
227 if isinstance(user, dict):
228 tenant = user.get("tenant_id") or user.get("tenant")
229 if isinstance(tenant, str):
230 tenant_id = tenant
231 gateway_request = RelayGatewayRequest(
232 request_id=request_id,
233 tenant_id=tenant_id,
234 source=RelayFormat.OPENAI_CHAT,
235 model=model_value,
236 stream=False,
237 payload=body,
238 headers=dict(request.headers.items()),
239 channel=None,
240 )
241 service = await resolve_passthrough(request)
242 result = await service.handle(kind, gateway_request)
243 if result.is_err():
244 return _error_response(RelayFormat.OPENAI_CHAT, result.unwrap_err())
245 ok_result = result.unwrap()
246 headers = _safe_headers(ok_result.headers, request_id, trace_id)
247 if ok_result.payload is not None:
248 return JSONResponse(
249 content=ok_result.payload,
250 status_code=ok_result.status_code,
251 headers=headers,
252 )
253 return Response(status_code=204, headers=headers)
256async def job_submit_endpoint(
257 kind: str,
258 resolve_job_passthrough: ResolveJobPassthrough,
259 request: Request,
260) -> Response:
261 """Serve one job-relay submit request in its own wire format.
263 The body is read exactly once and forwarded verbatim: no format
264 inference happens, so the request carries the conventional
265 ``OPENAI_CHAT`` source marker and no channel hint. Video is an
266 OpenAI-shaped submit/poll convention, so failures render in the
267 OpenAI error envelope through the same machinery as the passthrough
268 routes. The job passthrough service is resolved per request, never
269 cached.
271 Args:
272 kind: The endpoint kind owned by this route.
273 resolve_job_passthrough: Resolver of the job passthrough
274 service.
275 request: The Starlette request being served.
277 Returns:
278 The upstream JSON verbatim (with the id rewritten to the
279 gateway-issued job id), ``204`` when the result carries no
280 payload, or the OpenAI error envelope for gateway failures.
281 """
282 raw = await request.body()
283 request_id = getattr(request.state, "request_id", None) or new_uuid()
284 trace_id = request.headers.get("x-trace-id", "") or ""
285 body = _parse_body(raw, RelayFormat.OPENAI_CHAT, request_id)
286 if isinstance(body, Response):
287 return body
288 model_value = body.get("model")
289 if not isinstance(model_value, str) or not model_value:
290 return _error_response(
291 RelayFormat.OPENAI_CHAT,
292 RelayGatewayError(
293 code=RelayGatewayErrorCode.INVALID_REQUEST,
294 message="model is required",
295 status_code=400,
296 request_id=request_id,
297 ),
298 )
299 user = getattr(request.state, "user", None)
300 tenant_id = ""
301 if isinstance(user, dict):
302 tenant = user.get("tenant_id") or user.get("tenant")
303 if isinstance(tenant, str):
304 tenant_id = tenant
305 gateway_request = RelayGatewayRequest(
306 request_id=request_id,
307 tenant_id=tenant_id,
308 source=RelayFormat.OPENAI_CHAT,
309 model=model_value,
310 stream=False,
311 payload=body,
312 headers=dict(request.headers.items()),
313 channel=None,
314 )
315 service = await resolve_job_passthrough(request)
316 result = await service.submit(kind, gateway_request)
317 if result.is_err():
318 return _error_response(RelayFormat.OPENAI_CHAT, result.unwrap_err())
319 ok_result = result.unwrap()
320 headers = _safe_headers(ok_result.headers, request_id, trace_id)
321 if ok_result.payload is not None:
322 return JSONResponse(
323 content=ok_result.payload,
324 status_code=ok_result.status_code,
325 headers=headers,
326 )
327 return Response(status_code=204, headers=headers)
330async def job_status_endpoint(
331 kind: str,
332 resolve_job_passthrough: ResolveJobPassthrough,
333 request: Request,
334) -> Response:
335 """Serve one job-relay status poll against a gateway-issued job id.
337 The job id comes from the path, not the body: status polls carry no
338 payload, so the gateway request built here carries an empty payload
339 and model, and only identity and request id are meaningful. The
340 status call authorizes but never re-runs the billing pipeline, and
341 failures render in the OpenAI error envelope like the other
342 sighted routes.
344 Args:
345 kind: The endpoint kind owned by this route.
346 resolve_job_passthrough: Resolver of the job passthrough
347 service.
348 request: The Starlette request being served.
350 Returns:
351 The upstream status JSON verbatim (with the id rewritten back to
352 the gateway-issued job id), ``204`` when the result carries no
353 payload, or the OpenAI error envelope for gateway failures.
354 """
355 request_id = getattr(request.state, "request_id", None) or new_uuid()
356 trace_id = request.headers.get("x-trace-id", "") or ""
357 job_id = request.path_params.get("job_id")
358 if not isinstance(job_id, str) or not job_id:
359 return _error_response(
360 RelayFormat.OPENAI_CHAT,
361 RelayGatewayError(
362 code=RelayGatewayErrorCode.INVALID_REQUEST,
363 message="job_id is required",
364 status_code=400,
365 request_id=request_id,
366 ),
367 )
368 user = getattr(request.state, "user", None)
369 tenant_id = ""
370 if isinstance(user, dict):
371 tenant = user.get("tenant_id") or user.get("tenant")
372 if isinstance(tenant, str):
373 tenant_id = tenant
374 gateway_request = RelayGatewayRequest(
375 request_id=request_id,
376 tenant_id=tenant_id,
377 source=RelayFormat.OPENAI_CHAT,
378 model="",
379 stream=False,
380 payload={},
381 headers=dict(request.headers.items()),
382 channel=None,
383 )
384 service = await resolve_job_passthrough(request)
385 result = await service.status(kind, job_id, gateway_request)
386 if result.is_err():
387 return _error_response(RelayFormat.OPENAI_CHAT, result.unwrap_err())
388 ok_result = result.unwrap()
389 headers = _safe_headers(ok_result.headers, request_id, trace_id)
390 if ok_result.payload is not None:
391 return JSONResponse(
392 content=ok_result.payload,
393 status_code=ok_result.status_code,
394 headers=headers,
395 )
396 return Response(status_code=204, headers=headers)
399def build_routes(
400 resolve_gateway: ResolveGateway,
401 *,
402 resolve_passthrough: ResolvePassthrough | None = None,
403 resolve_job_passthrough: ResolveJobPassthrough | None = None,
404) -> list[Route]:
405 """Build the relay POST routes bound to gateway resolvers.
407 Args:
408 resolve_gateway: Async callable resolving a ``RelayGatewayProtocol``
409 from the request; wired to request-time DI by the contributor.
410 resolve_passthrough: Optional async callable resolving a
411 ``PassthroughService`` from the request; when provided, the
412 passthrough routes (e.g. ``/v1/embeddings``), the audio
413 routes (``/v1/audio/*``), and the image routes
414 (``/v1/images/*``) are appended.
415 resolve_job_passthrough: Optional async callable resolving a
416 ``JobPassthroughService`` from the request; when provided,
417 the job-relay routes (``POST /v1/videos`` and
418 ``GET /v1/videos/{job_id}``) are appended.
420 Returns:
421 One ``Route`` per inbound relay format, in ``RELAY_ROUTE_PATHS``
422 order, followed by the passthrough, audio, and image routes when
423 their resolver is provided and the job-relay routes when theirs
424 is.
425 """
426 routes = [
427 Route(
428 path,
429 partial(relay_endpoint, source, resolve_gateway),
430 methods=["POST"],
431 )
432 for path, source in _ROUTE_TABLE
433 ]
434 if resolve_passthrough is not None:
435 routes.extend(
436 Route(
437 path,
438 partial(passthrough_endpoint, kind, resolve_passthrough),
439 methods=["POST"],
440 )
441 for path, kind in _PASSTHROUGH_ROUTE_TABLE
442 )
443 routes.extend(
444 Route(
445 path,
446 partial(_AUDIO_HANDLERS[kind], resolve_passthrough),
447 methods=["POST"],
448 )
449 for path, kind in AUDIO_ROUTE_TABLE
450 )
451 routes.extend(build_image_routes(resolve_passthrough))
452 if resolve_job_passthrough is not None:
453 routes.extend(
454 Route(
455 path,
456 partial(job_submit_endpoint, kind, resolve_job_passthrough),
457 methods=["POST"],
458 )
459 for path, kind in _JOB_ROUTE_TABLE
460 )
461 routes.extend(
462 Route(
463 _JOB_STATUS_PATH,
464 partial(job_status_endpoint, kind, resolve_job_passthrough),
465 methods=["GET"],
466 )
467 for _, kind in _JOB_ROUTE_TABLE
468 )
469 return routes
472def _streaming_response(
473 source: RelayFormat,
474 stream: AsyncIterator[RelayWireEvent],
475 headers: dict[str, str],
476) -> StreamingResponse:
477 """Build the SSE response, framing events in the client's protocol.
479 Args:
480 source: The client's wire format; frames follow its syntax.
481 stream: The gateway's normalized event stream.
482 headers: Safe headers; ``cache-control`` and ``connection`` are
483 added here.
485 Returns:
486 A ``text/event-stream`` streaming response over the framed
487 events, with the protocol terminator emitted exactly once.
488 """
489 headers["cache-control"] = "no-cache"
490 headers["connection"] = "keep-alive"
491 encoder = SSEEncoder(source)
493 async def frames() -> AsyncIterator[bytes]:
494 terminal_event: RelayWireEvent | None = None
495 async for event in stream:
496 if event.terminal:
497 terminal_event = event
498 yield encoder.encode(event)
499 if event.terminal:
500 break
501 final = encoder.encode_terminal(source, terminal_event)
502 if final:
503 yield final
505 return StreamingResponse(frames(), media_type="text/event-stream", headers=headers)