Coverage for src/lexigram/web/docs/generator.py: 7%
185 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 04:37 +0800
1"""OpenAPI specification generator for lexigram-web
3This is a lightweight implementation that inspects controller route metadata
4and constructs a minimal OpenAPI 3.0.3 specification including component
5schemas for Pydantic response models (when available).
6"""
8from __future__ import annotations
10import inspect
11import re
12from typing import Any
14from lexigram.logging import get_logger
16logger = get_logger(__name__)
19class OpenAPIGenerator:
20 """Lightweight OpenAPI 3.0.3 specification generator.
22 Inspects controller route metadata collected by the routing layer and
23 produces a minimal OpenAPI spec dict that can be served as JSON at
24 ``/openapi.json`` or rendered via Swagger/Redoc UI.
26 Supports Pydantic response models (``$ref`` component schemas), built-in
27 Python types (inline schemas), and ``list[Model]`` generic aliases.
29 Register through the :class:`~lexigram.web.di.provider.WebProvider` so
30 the spec is generated lazily from the final route table::
32 # Via WebProvider (automatic)
33 # The provider calls OpenAPIGenerator(title=config.title, ...) and
34 # mounts /openapi.json and /docs routes automatically.
36 Args:
37 title: API title displayed in generated documentation.
38 version: API version string (e.g. ``"1.0.0"``).
39 description: Optional long-form API description shown in the UI.
40 """
42 def __init__(
43 self,
44 title: str,
45 version: str,
46 description: str | None = None,
47 ) -> None:
48 """
49 Args:
50 title: API title displayed in generated documentation.
51 version: API version string (e.g. ``"1.0.0"``).
52 description: Optional long-form API description shown in the UI.
53 """
54 self.title = title
55 self.version = version
56 self.description = description or ""
57 self.paths: dict[str, Any] = {}
58 self.components: dict[str, Any] = {"schemas": {}}
60 def _get_schema_for_model(self, model: Any) -> dict[str, Any]:
61 """Get an OpenAPI schema for a model or type.
63 Handles Pydantic models (with components/$ref), built-in types (inline),
64 and GenericAliases (like list[Model]).
65 """
66 if model is None:
67 return {"type": "object"}
69 # Handle basic built-in types
70 if model is str:
71 return {"type": "string"}
72 if model is int:
73 return {"type": "integer"}
74 if model is float:
75 return {"type": "number"}
76 if model is bool:
77 return {"type": "boolean"}
78 if model is list:
79 return {"type": "array", "items": {"type": "object"}}
80 if model is dict:
81 return {"type": "object"}
83 # Handle GenericAlias (list[Breed], dict[str, Any])
84 origin = getattr(model, "__origin__", None)
85 if origin is list:
86 args = getattr(model, "__args__", [])
87 items = self._get_schema_for_model(args[0]) if args else {"type": "object"}
88 return {"type": "array", "items": items}
89 if origin is dict:
90 return {"type": "object"}
92 # Handle Pydantic models (V2)
93 if hasattr(model, "model_json_schema"):
94 name = getattr(model, "__name__", "AnonymousModel")
95 if name not in self.components["schemas"]:
96 try:
97 schema = model.model_json_schema()
98 # If there is a $defs section, merge it into components
99 if "$defs" in schema:
100 defs = schema.pop("$defs")
101 for def_name, def_schema in defs.items():
102 self.components["schemas"][def_name] = def_schema
103 self.components["schemas"][name] = schema
104 except (TypeError, ValueError, AttributeError) as e:
105 logger.warning("Failed to generate JSON schema for %s: %s", name, e)
106 self.components["schemas"][name] = {"type": "object"}
107 return {"$ref": f"#/components/schemas/{name}"}
109 # Fallback for other classes with __name__
110 name = getattr(model, "__name__", None)
111 if name:
112 if name not in self.components["schemas"]:
113 self.components["schemas"][name] = {"type": "object"}
114 return {"$ref": f"#/components/schemas/{name}"}
116 return {"type": "object"}
118 def _normalize_path(self, path: Any) -> str:
119 """Ensures path is a string and replaces path parameters with OpenAPI format."""
120 if not isinstance(path, (str, bytes)):
121 path = str(path)
122 # Strip Starlette-style type hints from path parameters: {param:type} -> {param}
123 return re.sub(r"{([^}:]+):[^}]+}", r"{\1}", path)
125 def _extract_path_parameters(self, path: Any) -> list[dict[str, Any]]:
126 path = str(path)
127 params = []
128 for match in re.finditer(r"{([^}]+)}", path):
129 name = match.group(1).split(":")[0]
130 params.append(
131 {
132 "name": name,
133 "in": "path",
134 "required": True,
135 "schema": {"type": "string"},
136 },
137 )
138 return params
140 def _extract_operation_parameters(
141 self,
142 handler: Any,
143 path_params: list[dict[str, Any]],
144 ) -> list[dict[str, Any]]:
145 """Extract parameters from handler metadata and signature."""
146 params = list(path_params)
148 # 1. From _param_metadata (decorators like @query(func))
149 metadata = list(getattr(handler, "_param_metadata", []))
151 # 2. From signature defaults (decorators like p: str = header(...))
152 try:
153 sig = inspect.signature(handler)
154 for name, param in sig.parameters.items():
155 info = getattr(param.default, "_lexigram_param_info", None)
156 if info:
157 # Avoid duplicates if name already in metadata
158 if not any(m.get("name") == name for m in metadata):
159 meta = info.copy()
160 if not meta.get("name"):
161 meta["name"] = name
162 # Try to get annotation from type hints
163 try:
164 from typing import get_type_hints
166 hints = get_type_hints(handler)
167 meta["annotation"] = hints.get(name)
168 except (TypeError, NameError, AttributeError):
169 meta["annotation"] = param.annotation
170 metadata.append(meta)
171 except (ValueError, TypeError):
172 pass
174 for meta in metadata:
175 p_type = meta.get("type")
176 if p_type in ("query", "header", "cookie", "form"):
177 name = meta.get("alias") or meta.get("name")
178 if not name:
179 continue
181 # Check if already added
182 if any(p["name"] == name and p.get("in") == p_type for p in params):
183 continue
185 if p_type == "form":
186 p = {
187 "name": name,
188 "in": "form",
189 "required": meta.get("default") is ...,
190 "schema": {"type": "string"},
191 }
192 else:
193 p = {
194 "name": name,
195 "in": p_type,
196 "required": meta.get("default") is ...,
197 "schema": {"type": "string"}, # Default to string
198 }
200 # Try to refine schema from annotation if available
201 annotation = meta.get("annotation")
202 if annotation and annotation is not inspect.Parameter.empty:
203 from lexigram.web.docs.type_registry import (
204 _type_documenter_registry,
205 )
207 documenter = _type_documenter_registry.get_documenter(annotation)
208 if documenter:
209 documenter.document(p)
211 params.append(p)
212 elif p_type == "path":
213 # Path params are already extracted from the URL, but we can refine them
214 name = meta.get("alias") or meta.get("name")
215 if name:
216 for p in params:
217 if p["name"] == name:
218 annotation = meta.get("annotation")
219 if annotation and annotation is not inspect.Parameter.empty:
220 from lexigram.web.docs.type_registry import (
221 _type_documenter_registry,
222 )
224 documenter = _type_documenter_registry.get_documenter(
225 annotation,
226 )
227 if documenter:
228 documenter.document(p)
229 break
231 return params
233 def generate_spec(self, controllers: list[type]) -> dict[str, Any]:
234 """Generate an OpenAPI 3.0.3 specification for the provided controllers."""
235 for controller in controllers or []:
236 try:
237 from typing import cast
238 routes = cast("Any", controller).collect_routes()
239 except (AttributeError, TypeError) as e:
240 logger.warning(
241 "Controller %s collect_routes failed: %s",
242 getattr(controller, "__name__", repr(controller)),
243 e,
244 )
245 routes = []
247 tag = getattr(controller, "__name__", "default")
248 prefix = getattr(controller, "prefix", "") or ""
250 for route in routes:
251 raw_path = route.get("path")
252 if raw_path is None:
253 continue
255 # Harden path and method against Mocks
256 raw_path = str(raw_path)
257 # Ensure path starts with / and combines prefix correctly
258 path = prefix.rstrip("/") + raw_path
259 if not path.startswith("/"):
260 path = "/" + path
261 method = str(route.get("method") or "GET").lower()
263 handler_name = route.get("handler_name")
264 handler = (
265 getattr(controller, handler_name, None) if handler_name else None
266 )
268 # Extract summary and description from docstring if not provided
269 doc_summary = None
270 doc_description = None
271 if handler and handler.__doc__:
272 doc_lines = handler.__doc__.strip().split("\n")
273 doc_summary = doc_lines[0].strip()
274 if len(doc_lines) > 1:
275 doc_description = "\n".join(doc_lines[1:]).strip()
277 summary = route.get("summary") or doc_summary or handler_name
278 description = route.get("description") or doc_description or ""
280 op_id = route.get("operation_id") or handler_name
281 operation: dict[str, Any] = {
282 "operationId": f"{tag}_{op_id}" if op_id else None,
283 "summary": summary,
284 "description": description,
285 "tags": route.get("tags") or [tag],
286 "deprecated": route.get("deprecated", False),
287 "responses": {},
288 }
290 # Handle path parameters
291 raw_path_params = self._extract_path_parameters(path)
292 # Separate OpenAPI parameters from form-based body parameters
293 all_params = self._extract_operation_parameters(
294 handler,
295 raw_path_params,
296 )
297 operation["parameters"] = [p for p in all_params if p["in"] != "form"]
298 form_params = [p for p in all_params if p["in"] == "form"]
300 # Handle responses
301 status_code = str(route.get("status_code", 200))
302 custom_responses = route.get("responses")
304 if custom_responses:
305 for code, resp_data in custom_responses.items():
306 operation["responses"][str(code)] = resp_data
307 else:
308 operation["responses"][status_code] = {"description": "Success"}
310 # Request body (if a request model is provided or we have form params)
311 req_model = route.get("request_model")
312 if req_model is not None:
313 schema = self._get_schema_for_model(req_model)
314 operation["requestBody"] = {
315 "required": True,
316 "content": {
317 "application/json": {
318 "schema": schema,
319 },
320 },
321 }
322 elif form_params:
323 # Collect form params into a schema
324 form_properties = {}
325 required_form_params = []
326 for f_p in form_params:
327 form_properties[f_p["name"]] = f_p.get(
328 "schema", {"type": "string"},
329 )
330 if f_p.get("required"):
331 required_form_params.append(f_p["name"])
333 operation["requestBody"] = {
334 "content": {
335 "application/x-www-form-urlencoded": {
336 "schema": {
337 "type": "object",
338 "properties": form_properties,
339 },
340 },
341 },
342 }
343 if required_form_params:
344 operation["requestBody"]["content"][
345 "application/x-www-form-urlencoded"
346 ]["schema"]["required"] = required_form_params
348 resp_model = route.get("response_model")
349 if resp_model is not None:
350 schema = self._get_schema_for_model(resp_model)
351 # Merge with existing response if already defined via custom_responses
352 if status_code not in operation["responses"]:
353 operation["responses"][status_code] = {"description": "Success"}
355 operation["responses"][status_code]["content"] = {
356 "application/json": {
357 "schema": schema,
358 },
359 }
361 # Add path/method
362 path = self._normalize_path(path)
363 method = str(method).lower()
364 self.paths.setdefault(path, {})[method] = operation
366 spec: dict[str, Any] = {
367 "openapi": "3.0.3",
368 "info": {
369 "title": self.title,
370 "version": self.version,
371 "description": self.description,
372 },
373 "paths": self.paths,
374 "components": self.components,
375 }
376 return spec