« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Served-model catalog for the gateway's ``/v1/models`` surface.
3``ModelCatalogService`` aggregates the client-visible model aliases from
4the configured channel table into the three wire formats' list and
5detail shapes (OpenAI, Anthropic, Gemini). It is a pure function of the
6channel registry's static table and runtime overrides: drained and
7config-disabled channels contribute nothing, aliases are deduplicated,
8and the payloads never leak upstream URLs or credentials.
9"""
11from __future__ import annotations
13from typing import Any
15from lexigram.ai.relay.gateway.channels import RelayChannelRegistry
17__all__ = ["ModelCatalogService"]
19_OWNED_BY = "lexigram"
20"""Owner label reported in OpenAI model list entries."""
22_CREATED_AT = "1970-01-01T00:00:00Z"
23"""Fixed created-at stamp reported in Anthropic model entries."""
25_GENERATION_METHODS = ("generateContent",)
26"""Generation methods reported for Gemini model entries."""
29class ModelCatalogService:
30 """Aggregate served model aliases per wire format.
32 Args:
33 registry: The channel registry whose enabled, non-drained
34 channels define the served model set.
35 """
37 def __init__(self, registry: RelayChannelRegistry) -> None:
38 """Bind the catalog to the channel registry.
40 Args:
41 registry: The channel registry backing the model set.
42 """
43 self._registry = registry
45 def list_openai(self) -> dict[str, Any]:
46 """Return the OpenAI ``/v1/models`` list payload.
48 Returns:
49 A list payload with one entry per served alias, sorted.
50 """
51 return {
52 "object": "list",
53 "data": [
54 {
55 "id": alias,
56 "object": "model",
57 "created": 0,
58 "owned_by": _OWNED_BY,
59 }
60 for alias in self._served_models()
61 ],
62 }
64 def list_claude(self) -> dict[str, Any]:
65 """Return the Anthropic ``/v1/models`` list payload.
67 Returns:
68 A list payload with one entry per served alias, sorted.
69 """
70 return {
71 "data": [
72 {
73 "type": "model",
74 "id": alias,
75 "display_name": alias,
76 "created_at": _CREATED_AT,
77 }
78 for alias in self._served_models()
79 ],
80 }
82 def list_gemini(self) -> dict[str, Any]:
83 """Return the Gemini ``/v1beta/models`` list payload.
85 Returns:
86 A list payload with one model entry per served alias, sorted.
87 """
88 return {
89 "models": [
90 {
91 "name": f"models/{alias}",
92 "displayName": alias,
93 "supportedGenerationMethods": list(_GENERATION_METHODS),
94 }
95 for alias in self._served_models()
96 ],
97 }
99 def model_exists(self, alias: str) -> bool:
100 """Return whether *alias* is served by any enabled channel.
102 Args:
103 alias: The model alias to look up.
105 Returns:
106 ``True`` when the alias is served, ``False`` otherwise.
107 """
108 return alias in self._served_models()
110 def openai_detail(self, alias: str) -> dict[str, Any] | None:
111 """Return the OpenAI model detail payload for *alias*.
113 Args:
114 alias: The model alias to describe.
116 Returns:
117 The detail payload, or ``None`` when the alias is not served.
118 """
119 if not self.model_exists(alias):
120 return None
121 return {
122 "id": alias,
123 "object": "model",
124 "created": 0,
125 "owned_by": _OWNED_BY,
126 }
128 def gemini_detail(self, alias: str) -> dict[str, Any] | None:
129 """Return the Gemini model detail payload for *alias*.
131 Args:
132 alias: The model alias to describe.
134 Returns:
135 The detail payload, or ``None`` when the alias is not served.
136 """
137 if not self.model_exists(alias):
138 return None
139 return {
140 "name": f"models/{alias}",
141 "displayName": alias,
142 "supportedGenerationMethods": list(_GENERATION_METHODS),
143 }
145 def _served_models(self) -> tuple[str, ...]:
146 """Return the sorted, deduplicated served alias set."""
147 served: set[str] = set()
148 for channel in self._registry.channels:
149 if channel.enabled and self._registry.runtime_enabled().get(
150 channel.name, True
151 ):
152 served.update(channel.models)
153 return tuple(sorted(served))