src/genro_asgi/plugins/openapi/translator.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """OpenAPITranslator — turn genro-routes ``nodes()`` output into OpenAPI.16 17 Reads the dialect-neutral description produced by ``router.nodes()`` and NEVER18 re-derives anything from the callables. genro-routes' pydantic plugin computes19 and caches, once at decoration time, both the input and the output description20 of every handler; this translator only reads those neutral blocks:21 22 - input — ``entry_info["params"]``: ``schema`` is the aggregate request JSON23 schema (``request_schema``) and ``fields`` is the per-parameter list24 (``{name, schema, required, default, kind}``). The HTTP method is guessed25 from the per-parameter schemas (all scalar → GET, else POST), and the query26 parameters / request body are built from the cached ``schema`` — no pydantic27 model is ever built here;28 - output — ``entry_info["result"]``: ``{schema, media_type}`` (the return-type29 schema the pydantic plugin produced via ``TypeAdapter``);30 - ``security`` / ``x-requires`` come from the per-entry ``auth`` / ``env``31 plugin config.32 33 Consequently this module imports NO pydantic (and does not inspect callables):34 pydantic is a concern of genro-routes' pydantic plugin, which owns the schema35 derivation. The cached ``schema`` blocks keep ``$defs`` inline (self-contained36 per entry); this translator lifts them into a document-level pool, as OpenAPI37 expects — copying each block before popping ``$defs`` so the plugin's cache is38 never mutated.39 40 The translator methods are staticmethods by design (the granted builder-style41 exception, coding rule 4): pure data-logic functions over the neutral node42 description, holding no instance state.43 """44 45 from __future__ import annotations46 47 from typing import Any48 49 __all__ = ["OpenAPITranslator"]50 51 52 class OpenAPITranslator:53 """Translate router ``nodes()`` output to OpenAPI format.54 55 Modes:56 - ``openapi``: flat format, all paths merged into a single paths dict.57 - ``h_openapi``: hierarchical format preserving the router tree.58 """59 60 # JSON-schema primitive types serializable as query-string parameters.61 SCALAR_JSON_TYPES: set[str] = {"string", "integer", "number", "boolean", "null"}62 63 @staticmethod64 def translate_openapi(65 nodes_data: dict[str, Any],66 lazy: bool = False,67 path_prefix: str = "",68 ) -> dict[str, Any]:69 """Translate ``nodes()`` output to flat OpenAPI format."""70 paths: dict[str, Any] = {}71 all_defs: dict[str, Any] = {}72 73 entries = nodes_data.get("entries", {})74 for entry_name, entry_info in entries.items():75 path = f"{path_prefix}/{entry_name}" if path_prefix else f"/{entry_name}"76 path_item, defs = OpenAPITranslator.entry_info_to_openapi(entry_name, entry_info)77 paths[path] = path_item78 all_defs.update(defs)79 80 routers_data = nodes_data.get("routers", {})81 routers: dict[str, Any]82 if lazy:83 routers = dict(routers_data)84 else:85 for child_name, child_data in routers_data.items():86 child_prefix = f"{path_prefix}/{child_name}" if path_prefix else f"/{child_name}"87 child_openapi = OpenAPITranslator.translate_openapi(88 child_data, lazy=False, path_prefix=child_prefix89 )90 paths.update(child_openapi.get("paths", {}))91 if "$defs" in child_openapi:92 all_defs.update(child_openapi["$defs"])93 routers = {}94 95 result: dict[str, Any] = {"paths": paths}96 if all_defs:97 result["$defs"] = all_defs98 if routers:99 result["routers"] = routers100 return result101 102 @staticmethod103 def translate_h_openapi(104 nodes_data: dict[str, Any],105 lazy: bool = False,106 path_prefix: str = "",107 ) -> dict[str, Any]:108 """Translate ``nodes()`` output to hierarchical OpenAPI format."""109 paths: dict[str, Any] = {}110 all_defs: dict[str, Any] = {}111 112 entries = nodes_data.get("entries", {})113 for entry_name, entry_info in entries.items():114 path = f"{path_prefix}/{entry_name}" if path_prefix else f"/{entry_name}"115 path_item, defs = OpenAPITranslator.entry_info_to_openapi(entry_name, entry_info)116 paths[path] = path_item117 all_defs.update(defs)118 119 routers_data = nodes_data.get("routers", {})120 routers: dict[str, Any]121 if lazy:122 routers = dict(routers_data)123 else:124 routers = {}125 for child_name, child_data in routers_data.items():126 child_h_openapi = OpenAPITranslator.translate_h_openapi(child_data, lazy=False)127 if child_h_openapi:128 routers[child_name] = child_h_openapi129 if "$defs" in child_h_openapi:130 all_defs.update(child_h_openapi.pop("$defs"))131 132 result: dict[str, Any] = {133 "description": nodes_data.get("description"),134 "owner_doc": nodes_data.get("owner_doc"),135 }136 if paths:137 result["paths"] = paths138 if all_defs:139 result["$defs"] = all_defs140 if routers:141 result["routers"] = routers142 return result143 144 @staticmethod145 def entry_info_to_openapi(146 name: str, entry_info: dict[str, Any]147 ) -> tuple[dict[str, Any], dict[str, Any]]:148 """Convert an entry info dict to an OpenAPI path item.149 150 HTTP method priority: explicit openapi plugin config, else guessed from151 the per-parameter schemas. Input (query params / request body) is built152 from the cached ``params`` block; the response schema is read from the153 neutral ``result`` block. Nothing is derived from the callable.154 """155 doc = entry_info.get("doc", "")156 summary = doc.split("\n")[0] if doc else name157 metadata = entry_info.get("metadata", {})158 collected_defs: dict[str, Any] = {}159 160 params_block = entry_info.get("params") or {}161 request_schema = params_block.get("schema")162 fields = params_block.get("fields") or []163 164 openapi_config = metadata.get("plugin_config", {}).get("openapi", {})165 explicit_method = openapi_config.get("method")166 if explicit_method:167 http_method = explicit_method.lower()168 else:169 http_method = OpenAPITranslator.guess_http_method(fields)170 171 if openapi_config.get("summary"):172 summary = openapi_config["summary"]173 174 operation: dict[str, Any] = {175 "operationId": name,176 "summary": summary,177 }178 if openapi_config.get("description"):179 operation["description"] = openapi_config["description"]180 elif doc:181 operation["description"] = doc182 183 if openapi_config.get("deprecated"):184 operation["deprecated"] = True185 186 tags = openapi_config.get("tags")187 if tags:188 operation["tags"] = tags if isinstance(tags, list) else [tags]189 190 # Request body / query params: read the cached input schema (copy before191 # lifting $defs so the plugin's cached schema is never mutated).192 if request_schema is not None:193 schema = dict(request_schema)194 defs = schema.pop("$defs", None)195 if defs:196 collected_defs.update(defs)197 if http_method == "get":198 parameters = OpenAPITranslator.schema_to_parameters(schema)199 if parameters:200 operation["parameters"] = parameters201 else:202 operation["requestBody"] = {203 "required": True,204 "content": {"application/json": {"schema": schema}},205 }206 207 # Response schema: read from the neutral result block.208 result_block = entry_info.get("result") or {}209 response_schema = result_block.get("schema")210 media_type = result_block.get("media_type") or "application/json"211 if response_schema is not None:212 response_schema = dict(response_schema)213 if "$defs" in response_schema:214 collected_defs.update(response_schema.pop("$defs"))215 operation["responses"] = {216 "200": {217 "description": "Successful response",218 "content": {media_type: {"schema": response_schema}},219 }220 }221 222 if "responses" not in operation:223 operation["responses"] = {"200": {"description": "Successful response"}}224 225 # Security: explicit override, else derived from the auth plugin config.226 plugins = entry_info.get("plugins", {})227 explicit_security = openapi_config.get("security")228 if explicit_security is not None:229 operation["security"] = explicit_security230 else:231 auth_plugin = plugins.get("auth")232 if auth_plugin is not None:233 auth_config = auth_plugin.get("config", {})234 auth_rule = auth_config.get("rule", "")235 security_scheme = openapi_config.get("security_scheme", "BearerAuth")236 if auth_rule:237 operation["security"] = [{security_scheme: []}]238 else:239 operation["security"] = []240 241 # x-requires from the env plugin config.242 env_plugin = plugins.get("env")243 if env_plugin is not None:244 env_config = env_plugin.get("config", {})245 env_requires = env_config.get("requires", "")246 if env_requires:247 operation["x-requires"] = env_requires248 249 return {http_method: operation}, collected_defs250 251 @staticmethod252 def guess_http_method(fields: list[dict[str, Any]]) -> str:253 """Guess the HTTP method from the neutral parameter fields.254 255 GET when every typed parameter carries a scalar JSON schema, else POST.256 Reads only the cached ``fields`` (never the callable); an untyped257 parameter — ``schema`` is ``None`` (unannotated, or ``*args``/``**kwargs``)258 — does not force POST, matching "no typed params → GET".259 """260 for field in fields:261 schema = field.get("schema")262 if schema is None:263 continue264 if not OpenAPITranslator._is_scalar_schema(schema):265 return "post"266 return "get"267 268 @staticmethod269 def _is_scalar_schema(schema: Any) -> bool:270 """Return True if a JSON-schema fragment is a scalar (GET-friendly) type.271 272 A ``$ref`` or ``allOf`` (nested model) is non-scalar; an ``anyOf`` /273 ``oneOf`` union is scalar when every branch is (e.g. ``int | None``); a274 bare ``enum`` is scalar.275 """276 if not isinstance(schema, dict):277 return False278 if "$ref" in schema or "allOf" in schema:279 return False280 for combiner in ("anyOf", "oneOf"):281 options = schema.get(combiner)282 if options is not None:283 return all(OpenAPITranslator._is_scalar_schema(option) for option in options)284 schema_type = schema.get("type")285 if schema_type in OpenAPITranslator.SCALAR_JSON_TYPES:286 return True287 if schema_type is None and "enum" in schema:288 return True289 return False290 291 @staticmethod292 def schema_to_parameters(schema: dict[str, Any]) -> list[dict[str, Any]]:293 """Convert an aggregate request JSON schema to OpenAPI query parameters."""294 properties = schema.get("properties", {})295 required_fields = set(schema.get("required", []))296 parameters: list[dict[str, Any]] = []297 298 for prop_name, prop_schema in properties.items():299 parameters.append(300 {301 "name": prop_name,302 "in": "query",303 "required": prop_name in required_fields,304 "schema": prop_schema,305 }306 )307 return parameters