From b9eedbc30efa69a1ed8d515d5d23fdbe9f24da62 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Thu, 21 May 2026 20:56:27 +0200 Subject: [PATCH 1/2] Derive JSON Schema from Python type hints in OpenAPI output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit qh builds every route with a generic endpoint(request: Request) signature, so the wrapped function's parameters and return type are invisible to FastAPI's OpenAPI machinery: /openapi.json had empty {} request/response schemas and no components.schemas, and openapi-typescript produced no useful types. This derives a complete OpenAPI document from the wrapped functions' Python type hints — additive, with the request-handling path untouched: - qh/openapi.py: python_type_to_json_schema() converts Python type hints to JSON Schema (primitives, list/dict/tuple, Optional/Union incl. PEP 604, Literal, Any, dataclasses, TypedDicts incl. total=False, Pydantic models, NamedTuples, Enums). Named composites are registered in components.schemas and referenced by $ref; the recursion guard handles self-referential types. - enhance_openapi_schema() now fills requestBody, parameters (path/query), responses.200 and components.schemas. - install_enhanced_openapi() overrides app.openapi to serve the enriched doc at /openapi.json, with a defensive fallback to FastAPI's plain schema. - endpoint.py exposes the resolved param -> HTTP-location map; app.py surfaces it via inspect_routes — single source of truth for body/path/query classification. - mk_app(enhanced_openapi=True) installs it by default. Adds qh/tests/test_openapi_schema.py (27 tests). Closes #8 --- qh/__init__.py | 9 +- qh/app.py | 18 + qh/endpoint.py | 6 + qh/openapi.py | 598 +++++++++++++++++++++++++++++--- qh/tests/test_openapi_schema.py | 316 +++++++++++++++++ 5 files changed, 901 insertions(+), 46 deletions(-) create mode 100644 qh/tests/test_openapi_schema.py diff --git a/qh/__init__.py b/qh/__init__.py index 2708ced..57fc4c6 100644 --- a/qh/__init__.py +++ b/qh/__init__.py @@ -23,7 +23,12 @@ from qh.types import register_type, register_json_type, TypeRegistry # OpenAPI and client generation (Phase 3) -from qh.openapi import export_openapi, enhance_openapi_schema +from qh.openapi import ( + export_openapi, + enhance_openapi_schema, + install_enhanced_openapi, + python_type_to_json_schema, +) from qh.client import ( mk_client_from_openapi, mk_client_from_url, @@ -117,6 +122,8 @@ # OpenAPI & Client (Phase 3) "export_openapi", "enhance_openapi_schema", + "install_enhanced_openapi", + "python_type_to_json_schema", "mk_client_from_openapi", "mk_client_from_url", "mk_client_from_app", diff --git a/qh/app.py b/qh/app.py index 7de3fb0..d3305dc 100644 --- a/qh/app.py +++ b/qh/app.py @@ -32,6 +32,7 @@ def mk_app( use_conventions: bool = False, async_funcs: Optional[List[Union[str, Callable]]] = None, async_config: Optional[Union[Dict[str, Any], "TaskConfig"]] = None, + enhanced_openapi: bool = True, **kwargs, ) -> FastAPI: """ @@ -69,6 +70,12 @@ def mk_app( - TaskConfig object (applies to all async_funcs) - Dict mapping function names to TaskConfig objects + enhanced_openapi: Whether to serve an enhanced OpenAPI document at + ``/openapi.json`` — one with ``requestBody`` / ``responses`` / + ``components.schemas`` derived from each function's Python type + hints (see :mod:`qh.openapi`). Defaults to True; the enhancement is + additive and falls back to FastAPI's plain schema if it ever fails. + **kwargs: Additional FastAPI() constructor kwargs (if creating new app) Returns: @@ -254,6 +261,13 @@ def mk_app( if task_config and getattr(task_config, "create_task_endpoints", True): add_task_endpoints(app, func.__name__) + # Serve an OpenAPI document with full request/response JSON Schema derived + # from the wrapped functions' Python type hints. + if enhanced_openapi: + from qh.openapi import install_enhanced_openapi + + install_enhanced_openapi(app) + return app @@ -280,6 +294,10 @@ def inspect_routes(app: FastAPI) -> List[Dict[str, Any]]: # Include original function if available (for OpenAPI/client generation) if hasattr(route.endpoint, "_qh_original_func"): route_info["function"] = route.endpoint._qh_original_func + # Include the resolved param -> TransformSpec map (HTTP-location + # classification) for OpenAPI request/response schema generation. + if hasattr(route.endpoint, "_qh_param_specs"): + route_info["param_specs"] = route.endpoint._qh_param_specs routes.append(route_info) return routes diff --git a/qh/endpoint.py b/qh/endpoint.py index b7db4a3..9c6c71d 100644 --- a/qh/endpoint.py +++ b/qh/endpoint.py @@ -312,6 +312,12 @@ def task_wrapper(**kwargs): endpoint.__doc__ = func.__doc__ # Store original function for OpenAPI/client generation endpoint._qh_original_func = func # type: ignore + # Store the resolved param -> TransformSpec map (the single source of truth + # for which HTTP location each parameter is read from). OpenAPI generation + # (qh.openapi) consumes this to classify body/path/query parameters without + # re-deriving the logic. + endpoint._qh_param_specs = param_specs # type: ignore + endpoint._qh_route_config = route_config # type: ignore return endpoint diff --git a/qh/openapi.py b/qh/openapi.py index 5a9aa06..e6f20bd 100644 --- a/qh/openapi.py +++ b/qh/openapi.py @@ -1,29 +1,453 @@ """ -Enhanced OpenAPI generation for qh. - -Extends FastAPI's OpenAPI generation with metadata needed for bidirectional -Python ↔ HTTP transformation: - -- x-python-signature: Full function signature with defaults -- x-python-module: Module path for imports -- x-python-transformers: Type transformation metadata -- x-python-examples: Generated examples for testing +OpenAPI generation for qh — including JSON Schema derived from Python type hints. + +``qh`` builds every route with a generic ``endpoint(request: Request)`` signature +(see :mod:`qh.endpoint`): the wrapped function's real parameters and return type +are parsed *manually* from the request body, which makes them invisible to +FastAPI's own OpenAPI machinery. Out of the box, therefore, ``/openapi.json`` +lists routes and docstrings but emits empty ``{}`` request/response schemas and +no ``components.schemas``. + +This module closes that gap. It introspects each wrapped function's Python type +hints and derives a complete OpenAPI document: + +- ``requestBody`` — a JSON Schema object built from the JSON-body parameters, +- ``parameters`` — path/query parameters, +- ``responses`` — a JSON Schema for the return type, +- ``components.schemas`` — one named schema per dataclass / ``TypedDict`` / + Pydantic model / ``NamedTuple`` / ``Enum`` encountered, referenced via ``$ref``, +- ``x-python-*`` extensions — Python signature metadata for bidirectional + Python ↔ HTTP transformation. + +The whole thing is **additive**: the request-handling path is untouched, and +:func:`install_enhanced_openapi` falls back to FastAPI's plain schema if +enhancement ever raises — so a converter bug can never break ``/openapi.json``. + +Public surface: + +- :func:`python_type_to_json_schema` — the Python-type-hint → JSON Schema converter. +- :func:`enhance_openapi_schema` — full enhanced OpenAPI document for an app. +- :func:`export_openapi` — :func:`enhance_openapi_schema` plus optional file output. +- :func:`install_enhanced_openapi` — make an app serve the enhanced doc at + ``/openapi.json`` (and render it in ``/docs``). """ +import collections.abc +import dataclasses +import enum +import inspect +import json from typing import ( Any, + Callable, Dict, List, + Literal, Optional, - Callable, - get_type_hints, - get_origin, + Union, get_args, + get_origin, + get_type_hints, ) -import inspect + from fastapi import FastAPI from fastapi.openapi.utils import get_openapi +try: # PEP 604 unions (``int | None``) — Python 3.10+ + from types import UnionType # type: ignore +except ImportError: # pragma: no cover - Python < 3.10 + UnionType = None # type: ignore + +NoneType = type(None) + +#: Exact-match JSON Schema for Python primitives. ``bool`` is listed before +#: ``int`` only for readability — lookup is by exact type, so order is moot. +_PRIMITIVE_SCHEMAS: Dict[type, Dict[str, Any]] = { + bool: {"type": "boolean"}, + int: {"type": "integer"}, + float: {"type": "number"}, + str: {"type": "string"}, + bytes: {"type": "string", "format": "byte"}, + NoneType: {"type": "null"}, +} + +#: Generic origins treated as JSON arrays. +_ARRAY_ORIGINS = ( + list, + tuple, + set, + frozenset, + collections.abc.Sequence, + collections.abc.MutableSequence, + collections.abc.Set, + collections.abc.MutableSet, + collections.abc.Collection, + collections.abc.Iterable, +) + +#: Generic origins treated as JSON objects (free-form maps). +_MAPPING_ORIGINS = ( + dict, + collections.abc.Mapping, + collections.abc.MutableMapping, +) + + +# --------------------------------------------------------------------------- +# Python type → JSON Schema +# --------------------------------------------------------------------------- + + +def _ref(name: str) -> Dict[str, str]: + """A JSON Schema ``$ref`` into ``components.schemas``.""" + return {"$ref": f"#/components/schemas/{name}"} + + +def _safe_type_hints(obj: Any) -> Dict[str, Any]: + """Resolve type hints, degrading gracefully when annotations don't resolve. + + ``get_type_hints`` raises ``NameError`` for string annotations referring to + names not in the object's module globals (e.g. types defined inside a + function). In that case fall back to the raw ``__annotations__`` mapping — + the converter treats any leftover string as an unknown (empty) schema. + """ + try: + return get_type_hints(obj) + except Exception: + return dict(getattr(obj, "__annotations__", {}) or {}) + + +def _is_typed_dict(tp: Any) -> bool: + """Whether ``tp`` is a ``typing.TypedDict`` subclass.""" + # ``__required_keys__`` / ``__optional_keys__`` are the stable markers. + return ( + isinstance(tp, type) + and hasattr(tp, "__required_keys__") + and hasattr(tp, "__annotations__") + ) + + +def _is_namedtuple(tp: Any) -> bool: + """Whether ``tp`` is a ``typing.NamedTuple`` / ``collections.namedtuple``.""" + return ( + isinstance(tp, type) + and issubclass(tp, tuple) + and hasattr(tp, "_fields") + and hasattr(tp, "__annotations__") + ) + + +def _is_pydantic_model(tp: Any) -> bool: + """Whether ``tp`` is a Pydantic model (v2 ``model_json_schema`` or v1 ``schema``).""" + return isinstance(tp, type) and ( + hasattr(tp, "model_json_schema") or hasattr(tp, "schema") + ) and hasattr(tp, "__fields__") + + +def python_type_to_json_schema( + type_hint: Any, + schemas: Dict[str, Any], + *, + _stack: Optional[frozenset] = None, +) -> Dict[str, Any]: + """Convert a Python type hint to a JSON Schema fragment. + + Primitives, containers and unions are inlined. Named composite types — + dataclasses, ``TypedDict``\\ s, Pydantic models, ``NamedTuple``\\ s and + ``Enum``\\ s — are registered in ``schemas`` (the OpenAPI + ``components.schemas`` table) and returned as a ``$ref``, so the same type + used in several places is described once. + + Args: + type_hint: the Python type / annotation to convert. ``Any`` and + :data:`inspect.Parameter.empty` map to the empty schema ``{}`` + (matches anything); ``None`` / ``NoneType`` map to ``{"type": "null"}``. + schemas: the mutable ``components.schemas`` accumulator — composite + types encountered are added here, keyed by their class name. + _stack: internal — the set of composite type names currently being + built, used to break recursion on self-referential types. + + Returns: + a JSON Schema dict — an inline fragment, or ``{"$ref": ...}`` for a + named composite type. + + Examples: + >>> python_type_to_json_schema(int, {}) + {'type': 'integer'} + >>> python_type_to_json_schema(list[str], {}) + {'type': 'array', 'items': {'type': 'string'}} + >>> python_type_to_json_schema(Optional[int], {}) + {'anyOf': [{'type': 'integer'}, {'type': 'null'}]} + """ + stack = _stack if _stack is not None else frozenset() + + # Unknowns and unresolved string annotations → permissive empty schema. + if type_hint is inspect.Parameter.empty or type_hint is Any: + return {} + if isinstance(type_hint, str): + return {} + + # Exact primitive match (covers ``NoneType`` too). + if type_hint in _PRIMITIVE_SCHEMAS: + return dict(_PRIMITIVE_SCHEMAS[type_hint]) + if type_hint is None: # a bare ``None`` annotation means ``NoneType`` + return {"type": "null"} + + origin = get_origin(type_hint) + args = get_args(type_hint) + + # Unions (``typing.Union`` and PEP 604 ``X | Y``), incl. ``Optional``. + if origin is Union or (UnionType is not None and origin is UnionType): + non_none = [a for a in args if a is not NoneType] + sub = [python_type_to_json_schema(a, schemas, _stack=stack) for a in non_none] + if len(non_none) != len(args): # had ``None`` → nullable + sub.append({"type": "null"}) + return sub[0] if len(sub) == 1 else {"anyOf": sub} + + # ``Literal[...]`` → enum of the literal values. + if origin is Literal: + return {"enum": list(args)} + + # Arrays. + if origin in _ARRAY_ORIGINS: + item_args = [a for a in args if a is not Ellipsis] + item = ( + python_type_to_json_schema(item_args[0], schemas, _stack=stack) + if item_args + else {} + ) + return {"type": "array", "items": item} + + # Free-form objects (maps). JSON object keys are always strings, so the key + # type is ignored; only the value type informs ``additionalProperties``. + if origin in _MAPPING_ORIGINS: + value = ( + python_type_to_json_schema(args[1], schemas, _stack=stack) + if len(args) == 2 + else {} + ) + return {"type": "object", "additionalProperties": value} + + # Bare ``list`` / ``dict`` (no parameters). + if type_hint is list: + return {"type": "array", "items": {}} + if type_hint is dict: + return {"type": "object", "additionalProperties": {}} + + # Named composite types → registered in ``schemas``, returned as ``$ref``. + if _is_typed_dict(type_hint): + return _register_object(type_hint, schemas, stack, _typed_dict_fields) + if dataclasses.is_dataclass(type_hint) and isinstance(type_hint, type): + return _register_object(type_hint, schemas, stack, _dataclass_fields) + if _is_pydantic_model(type_hint): + return _register_pydantic(type_hint, schemas) + if _is_namedtuple(type_hint): + return _register_object(type_hint, schemas, stack, _namedtuple_fields) + if isinstance(type_hint, type) and issubclass(type_hint, enum.Enum): + return {"enum": [member.value for member in type_hint]} + + # Primitive subclasses (e.g. ``class UserId(str)``). + if isinstance(type_hint, type): + for prim, prim_schema in _PRIMITIVE_SCHEMAS.items(): + if prim is not NoneType and issubclass(type_hint, prim): + return dict(prim_schema) + + # Unknown — permissive empty schema. + return {} + + +def _register_object( + tp: type, + schemas: Dict[str, Any], + stack: frozenset, + fields_fn: Callable[[type], List[tuple]], +) -> Dict[str, str]: + """Register an object-shaped composite type and return a ``$ref`` to it. + + ``fields_fn`` yields ``(name, type_hint, required)`` triples. The type name + is added to the recursion ``stack`` *before* its fields are converted, so a + self-referential field resolves to a ``$ref`` instead of recursing forever. + """ + name = tp.__name__ + if name in schemas or name in stack: + return _ref(name) + + child_stack = stack | {name} + properties: Dict[str, Any] = {} + required: List[str] = [] + for field_name, field_type, is_required in fields_fn(tp): + properties[field_name] = python_type_to_json_schema( + field_type, schemas, _stack=child_stack + ) + if is_required: + required.append(field_name) + + obj: Dict[str, Any] = {"type": "object", "properties": properties} + if required: + obj["required"] = required + doc = inspect.getdoc(tp) + if doc: + obj["description"] = doc + schemas[name] = obj + return _ref(name) + + +def _typed_dict_fields(tp: type) -> List[tuple]: + """``(name, type, required)`` triples for a ``TypedDict``. + + Honours ``total=False`` and per-key ``Required`` / ``NotRequired`` via the + ``__required_keys__`` / ``__optional_keys__`` markers. + """ + hints = _safe_type_hints(tp) + required_keys = set(getattr(tp, "__required_keys__", set())) + return [ + (key, hints.get(key, Any), key in required_keys) for key in hints + ] + + +def _dataclass_fields(tp: type) -> List[tuple]: + """``(name, type, required)`` triples for a dataclass — required = no default.""" + hints = _safe_type_hints(tp) + triples = [] + for field in dataclasses.fields(tp): + has_default = ( + field.default is not dataclasses.MISSING + or field.default_factory is not dataclasses.MISSING # type: ignore[misc] + ) + triples.append((field.name, hints.get(field.name, field.type), not has_default)) + return triples + + +def _namedtuple_fields(tp: type) -> List[tuple]: + """``(name, type, required)`` triples for a ``NamedTuple`` — required = no default.""" + hints = _safe_type_hints(tp) + defaults = getattr(tp, "_field_defaults", {}) + return [ + (name, hints.get(name, Any), name not in defaults) + for name in getattr(tp, "_fields", ()) + ] + + +def _register_pydantic(tp: type, schemas: Dict[str, Any]) -> Dict[str, str]: + """Register a Pydantic model via its own JSON Schema export, return a ``$ref``. + + Pydantic emits a self-contained schema; its nested-model definitions + (``$defs`` in v2, ``definitions`` in v1) are lifted into ``components.schemas``. + """ + name = tp.__name__ + if name in schemas: + return _ref(name) + if hasattr(tp, "model_json_schema"): # Pydantic v2 + doc = tp.model_json_schema(ref_template="#/components/schemas/{model}") + else: # Pydantic v1 + doc = tp.schema(ref_template="#/components/schemas/{model}") + defs = doc.pop("$defs", None) or doc.pop("definitions", None) or {} + schemas[name] = doc # reserve the name before recursing into nested defs + for def_name, def_schema in defs.items(): + schemas.setdefault(def_name, def_schema) + return _ref(name) + + +# --------------------------------------------------------------------------- +# Operation-level schema construction +# --------------------------------------------------------------------------- + + +def _param_location(param_specs: Dict[str, Any], param_name: str) -> str: + """The HTTP location (``json_body`` / ``path`` / ``query`` / …) of a parameter. + + Reads :class:`qh.rules.TransformSpec.http_location` from the resolved + ``param_specs`` map attached to the endpoint (the single source of truth). + Defaults to ``json_body`` when unknown. + """ + spec = param_specs.get(param_name) if param_specs else None + location = getattr(spec, "http_location", None) + return getattr(location, "value", "json_body") + + +def build_request_body_schema( + func: Callable, + param_specs: Dict[str, Any], + schemas: Dict[str, Any], +) -> Optional[Dict[str, Any]]: + """Build the JSON Schema object for a function's JSON-body parameters. + + Only parameters whose resolved HTTP location is the JSON body are included; + path/query/header parameters are emitted separately by + :func:`build_parameters`. A parameter with no default is ``required``. + + Returns: + an object JSON Schema, or ``None`` when the function has no body + parameters (e.g. a GET route whose arguments are all query parameters). + """ + sig = inspect.signature(func) + hints = _safe_type_hints(func) + + properties: Dict[str, Any] = {} + required: List[str] = [] + for name, param in sig.parameters.items(): + if _param_location(param_specs, name) != "json_body": + continue + properties[name] = python_type_to_json_schema(hints.get(name, Any), schemas) + if param.default is inspect.Parameter.empty: + required.append(name) + + if not properties: + return None + obj: Dict[str, Any] = {"type": "object", "properties": properties} + if required: + obj["required"] = required + return obj + + +def build_parameters( + func: Callable, + param_specs: Dict[str, Any], + schemas: Dict[str, Any], +) -> List[Dict[str, Any]]: + """Build OpenAPI ``parameters`` entries for a function's path/query arguments. + + Path parameters are always required; query parameters are required only + when the Python parameter has no default. + """ + sig = inspect.signature(func) + hints = _safe_type_hints(func) + + parameters: List[Dict[str, Any]] = [] + for name, param in sig.parameters.items(): + location = _param_location(param_specs, name) + if location not in ("path", "query"): + continue + spec = param_specs.get(name) + parameters.append( + { + "name": getattr(spec, "http_name", None) or name, + "in": location, + "required": location == "path" + or param.default is inspect.Parameter.empty, + "schema": python_type_to_json_schema(hints.get(name, Any), schemas), + } + ) + return parameters + + +def build_response_schema( + func: Callable, + schemas: Dict[str, Any], +) -> Dict[str, Any]: + """Build the JSON Schema for a function's return type. + + A missing return annotation yields the permissive empty schema; a ``None`` + return yields ``{"type": "null"}`` (``qh`` still replies with a JSON body). + """ + hints = _safe_type_hints(func) + return python_type_to_json_schema(hints.get("return", Any), schemas) + + +# --------------------------------------------------------------------------- +# Python signature metadata (x-python-* extensions) +# --------------------------------------------------------------------------- + def get_python_type_name(type_hint: Any) -> str: """ @@ -69,7 +493,7 @@ def extract_function_signature(func: Callable) -> Dict[str, Any]: - docstring: function docstring """ sig = inspect.signature(func) - type_hints = get_type_hints(func) if hasattr(func, "__annotations__") else {} + type_hints = _safe_type_hints(func) parameters = [] for param_name, param in sig.parameters.items(): @@ -109,7 +533,7 @@ def generate_examples_for_function(func: Callable) -> List[Dict[str, Any]]: Uses type hints to generate sensible example values. """ sig = inspect.signature(func) - type_hints = get_type_hints(func) if hasattr(func, "__annotations__") else {} + type_hints = _safe_type_hints(func) examples = [] @@ -170,26 +594,43 @@ def _generate_example_value(type_hint: Any, param_name: str) -> Any: return None +# --------------------------------------------------------------------------- +# Whole-document enhancement +# --------------------------------------------------------------------------- + + def enhance_openapi_schema( app: FastAPI, *, include_examples: bool = True, include_python_metadata: bool = True, + include_schemas: bool = True, include_transformers: bool = False, ) -> Dict[str, Any]: """ - Generate enhanced OpenAPI schema with Python-specific extensions. + Generate an enhanced OpenAPI schema for a ``qh`` app. + + On top of FastAPI's base document this fills in what ``qh``'s + ``Request``-based endpoints hide from FastAPI: + + - ``requestBody`` / ``parameters`` / ``responses`` JSON Schema derived from + each wrapped function's Python type hints (``include_schemas``), + - ``components.schemas`` for every dataclass / ``TypedDict`` / Pydantic + model / ``NamedTuple`` / ``Enum`` referenced, + - ``x-python-signature`` metadata (``include_python_metadata``), + - request examples (``include_examples``). Args: - app: FastAPI application - include_examples: Add example requests/responses - include_python_metadata: Add x-python-* extensions - include_transformers: Add transformation metadata + app: the FastAPI application. + include_examples: add example requests. + include_python_metadata: add ``x-python-*`` extensions. + include_schemas: derive ``requestBody`` / ``responses`` / + ``components.schemas`` from the Python type hints. + include_transformers: add (placeholder) transformation metadata. Returns: - Enhanced OpenAPI schema dictionary + the enhanced OpenAPI schema dictionary. """ - # Get base OpenAPI schema from FastAPI schema = get_openapi( title=app.title, version=app.version, @@ -198,12 +639,11 @@ def enhance_openapi_schema( routes=app.routes, ) - # Store function metadata by operation_id from qh.app import inspect_routes routes = inspect_routes(app) + component_schemas: Dict[str, Any] = {} - # Enhance each endpoint with Python metadata for route_info in routes: func = route_info.get("function") if not func: @@ -211,40 +651,37 @@ def enhance_openapi_schema( path = route_info["path"] methods = route_info.get("methods", ["POST"]) + param_specs = route_info.get("param_specs", {}) or {} - # Skip OpenAPI/docs routes if path in ["/openapi.json", "/docs", "/redoc"]: continue - # Find the operation in the schema path_item = schema.get("paths", {}).get(path, {}) for method in methods: method_lower = method.lower() operation = path_item.get(method_lower, {}) - if not operation: continue - # Add Python metadata if include_python_metadata: - sig_info = extract_function_signature(func) - operation["x-python-signature"] = sig_info + operation["x-python-signature"] = extract_function_signature(func) + + if include_schemas: + _apply_operation_schemas( + operation, func, param_specs, method_lower, component_schemas + ) - # Add examples if include_examples: examples = generate_examples_for_function(func) if examples and "requestBody" in operation: - content = operation["requestBody"].get("content", {}) - json_content = content.get("application/json", {}) + content = operation["requestBody"].setdefault("content", {}) + json_content = content.setdefault("application/json", {}) json_content["examples"] = { f"example_{i}": ex for i, ex in enumerate(examples) } - # Add transformer metadata (if requested) if include_transformers: - # This would include information about how types are transformed - # For now, we'll add a placeholder operation["x-python-transformers"] = { "note": "Type transformation metadata would go here" } @@ -253,29 +690,63 @@ def enhance_openapi_schema( schema["paths"][path] = path_item + if include_schemas and component_schemas: + components = schema.setdefault("components", {}) + components.setdefault("schemas", {}).update(component_schemas) + return schema +def _apply_operation_schemas( + operation: Dict[str, Any], + func: Callable, + param_specs: Dict[str, Any], + method_lower: str, + component_schemas: Dict[str, Any], +) -> None: + """Fill an operation's ``requestBody`` / ``parameters`` / ``responses`` in place.""" + parameters = build_parameters(func, param_specs, component_schemas) + if parameters: + existing = operation.get("parameters", []) + operation["parameters"] = existing + parameters + + if method_lower in ("post", "put", "patch"): + body = build_request_body_schema(func, param_specs, component_schemas) + if body is not None: + request_body = operation.setdefault("requestBody", {}) + request_body["required"] = bool(body.get("required")) + content = request_body.setdefault("content", {}) + content["application/json"] = {"schema": body} + + response = build_response_schema(func, component_schemas) + responses = operation.setdefault("responses", {}) + ok = responses.setdefault("200", {"description": "Successful Response"}) + ok.setdefault("content", {})["application/json"] = {"schema": response} + + def export_openapi( app: FastAPI, *, include_examples: bool = True, include_python_metadata: bool = True, + include_schemas: bool = True, include_transformers: bool = False, output_file: Optional[str] = None, ) -> Dict[str, Any]: """ - Export enhanced OpenAPI schema. + Export the enhanced OpenAPI schema, optionally writing it to a file. Args: - app: FastAPI application - include_examples: Include example requests/responses - include_python_metadata: Include x-python-* extensions - include_transformers: Include transformation metadata - output_file: Optional file path to write JSON output + app: the FastAPI application. + include_examples: include example requests. + include_python_metadata: include ``x-python-*`` extensions. + include_schemas: derive ``requestBody`` / ``responses`` / + ``components.schemas`` from the Python type hints. + include_transformers: include transformation metadata. + output_file: optional path to write the JSON document to. Returns: - Enhanced OpenAPI schema dictionary + the enhanced OpenAPI schema dictionary. Example: >>> from qh import mk_app # doctest: +SKIP @@ -287,13 +758,50 @@ def export_openapi( app, include_examples=include_examples, include_python_metadata=include_python_metadata, + include_schemas=include_schemas, include_transformers=include_transformers, ) if output_file: - import json - with open(output_file, "w") as f: json.dump(schema, f, indent=2) return schema + + +def install_enhanced_openapi(app: FastAPI, **enhance_kwargs: Any) -> FastAPI: + """Make ``app`` serve the enhanced OpenAPI schema at its ``/openapi.json``. + + Overrides ``app.openapi`` so the document returned by FastAPI — and rendered + by ``/docs`` and ``/redoc`` — carries the request/response JSON Schema and + ``components.schemas`` derived from the wrapped functions' Python type hints. + + The override is **defensive**: if enhancement raises for any reason, it + falls back to FastAPI's plain schema, so a converter bug can never turn + ``/openapi.json`` into a 500. + + Args: + app: the FastAPI application (typically the result of :func:`qh.mk_app`). + **enhance_kwargs: forwarded to :func:`enhance_openapi_schema`. + + Returns: + the same ``app``, for chaining. + """ + + def custom_openapi() -> Dict[str, Any]: + if app.openapi_schema: + return app.openapi_schema + try: + app.openapi_schema = enhance_openapi_schema(app, **enhance_kwargs) + except Exception: + app.openapi_schema = get_openapi( + title=app.title, + version=app.version, + openapi_version=app.openapi_version, + description=app.description, + routes=app.routes, + ) + return app.openapi_schema + + app.openapi = custom_openapi # type: ignore[method-assign] + return app diff --git a/qh/tests/test_openapi_schema.py b/qh/tests/test_openapi_schema.py new file mode 100644 index 0000000..1bf24a9 --- /dev/null +++ b/qh/tests/test_openapi_schema.py @@ -0,0 +1,316 @@ +"""Tests for JSON Schema derived from Python type hints in OpenAPI output. + +Covers :func:`qh.openapi.python_type_to_json_schema` and the request/response +schema it feeds into the OpenAPI document — the capability that lets tools like +``openapi-typescript`` generate a typed client from ``qh``'s ``/openapi.json``. + +The composite types under test are defined at **module level** on purpose: +``get_type_hints`` resolves string annotations against an object's module +globals, so a dataclass / ``TypedDict`` defined inside a test method would not +resolve. +""" + +import enum +from dataclasses import dataclass, field +from typing import Literal, NamedTuple, Optional, TypedDict, Union + +import pytest + +from qh import mk_app, export_openapi, python_type_to_json_schema +from qh.openapi import enhance_openapi_schema + + +# -------------------------------------------------------------------------- +# Composite types used across the tests +# -------------------------------------------------------------------------- + + +@dataclass +class Point: + """A 2-D point.""" + + x: float + y: float + + +@dataclass +class Annotated: + """A point with an optional label and a defaulted weight.""" + + point: Point + label: Optional[str] = None + weight: float = 1.0 + + +class Color(enum.Enum): + RED = "red" + GREEN = "green" + + +class CorpusSummary(TypedDict): + """A fully-required TypedDict (total=True).""" + + corpus_id: str + n_segments: int + + +class PartialMeta(TypedDict, total=False): + """A TypedDict whose every key is optional (total=False).""" + + note: str + score: float + + +class Pair(NamedTuple): + left: int + right: int = 0 + + +@dataclass +class TreeNode: + """A self-referential dataclass — exercises the recursion guard.""" + + value: int + children: list["TreeNode"] = field(default_factory=list) + + +# -------------------------------------------------------------------------- +# python_type_to_json_schema — primitives, containers, unions +# -------------------------------------------------------------------------- + + +class TestPrimitivesAndContainers: + def test_primitives(self): + assert python_type_to_json_schema(int, {}) == {"type": "integer"} + assert python_type_to_json_schema(str, {}) == {"type": "string"} + assert python_type_to_json_schema(float, {}) == {"type": "number"} + assert python_type_to_json_schema(bool, {}) == {"type": "boolean"} + assert python_type_to_json_schema(type(None), {}) == {"type": "null"} + + def test_list(self): + assert python_type_to_json_schema(list[str], {}) == { + "type": "array", + "items": {"type": "string"}, + } + + def test_nested_list(self): + assert python_type_to_json_schema(list[list[float]], {}) == { + "type": "array", + "items": {"type": "array", "items": {"type": "number"}}, + } + + def test_dict(self): + assert python_type_to_json_schema(dict[str, int], {}) == { + "type": "object", + "additionalProperties": {"type": "integer"}, + } + + def test_bare_containers(self): + assert python_type_to_json_schema(list, {}) == { + "type": "array", + "items": {}, + } + assert python_type_to_json_schema(dict, {}) == { + "type": "object", + "additionalProperties": {}, + } + + def test_optional(self): + assert python_type_to_json_schema(Optional[int], {}) == { + "anyOf": [{"type": "integer"}, {"type": "null"}] + } + + def test_pep604_optional(self): + assert python_type_to_json_schema(int | None, {}) == { + "anyOf": [{"type": "integer"}, {"type": "null"}] + } + + def test_union(self): + schema = python_type_to_json_schema(Union[int, str], {}) + assert schema == {"anyOf": [{"type": "integer"}, {"type": "string"}]} + + def test_literal(self): + assert python_type_to_json_schema(Literal["a", "b"], {}) == { + "enum": ["a", "b"] + } + + def test_enum(self): + assert python_type_to_json_schema(Color, {}) == {"enum": ["red", "green"]} + + +# -------------------------------------------------------------------------- +# python_type_to_json_schema — composite types into components +# -------------------------------------------------------------------------- + + +class TestCompositeTypes: + def test_dataclass_registered_as_ref(self): + schemas = {} + result = python_type_to_json_schema(Point, schemas) + assert result == {"$ref": "#/components/schemas/Point"} + assert schemas["Point"]["type"] == "object" + assert schemas["Point"]["properties"]["x"] == {"type": "number"} + assert set(schemas["Point"]["required"]) == {"x", "y"} + assert schemas["Point"]["description"] == "A 2-D point." + + def test_dataclass_defaults_not_required(self): + schemas = {} + python_type_to_json_schema(Annotated, schemas) + annotated = schemas["Annotated"] + # Only ``point`` has no default → only ``point`` is required. + assert annotated["required"] == ["point"] + # The nested dataclass is registered too, and referenced. + assert annotated["properties"]["point"] == { + "$ref": "#/components/schemas/Point" + } + assert "Point" in schemas + + def test_typed_dict_total(self): + schemas = {} + python_type_to_json_schema(CorpusSummary, schemas) + summary = schemas["CorpusSummary"] + assert set(summary["required"]) == {"corpus_id", "n_segments"} + + def test_typed_dict_total_false(self): + schemas = {} + python_type_to_json_schema(PartialMeta, schemas) + partial = schemas["PartialMeta"] + # total=False → no key is required. + assert "required" not in partial + assert set(partial["properties"]) == {"note", "score"} + + def test_namedtuple(self): + schemas = {} + python_type_to_json_schema(Pair, schemas) + pair = schemas["Pair"] + # ``right`` has a default → not required. + assert pair["required"] == ["left"] + + def test_list_of_dataclass(self): + schemas = {} + result = python_type_to_json_schema(list[Point], schemas) + assert result == { + "type": "array", + "items": {"$ref": "#/components/schemas/Point"}, + } + assert "Point" in schemas + + def test_self_referential_dataclass_terminates(self): + schemas = {} + result = python_type_to_json_schema(TreeNode, schemas) + assert result == {"$ref": "#/components/schemas/TreeNode"} + children = schemas["TreeNode"]["properties"]["children"] + assert children == { + "type": "array", + "items": {"$ref": "#/components/schemas/TreeNode"}, + } + + def test_unknown_type_is_permissive(self): + # An unannotated / unknown type maps to the empty (matches-anything) schema. + assert python_type_to_json_schema(object, {}) == {} + + +# -------------------------------------------------------------------------- +# Whole-document enhancement +# -------------------------------------------------------------------------- + + +def make_point(x: float, y: float) -> Point: + """Create a point from coordinates.""" + return Point(x, y) + + +def summarize(label: str, count: int = 0) -> CorpusSummary: + """Summarize.""" + return CorpusSummary(corpus_id=label, n_segments=count) + + +def get_item(item_id: str) -> Point: + """Fetch an item by id (convention-routed GET → query param).""" + return Point(0.0, 0.0) + + +class TestEnhancedDocument: + def test_request_body_schema(self): + app = mk_app([make_point]) + spec = export_openapi(app) + body = spec["paths"]["/make_point"]["post"]["requestBody"] + schema = body["content"]["application/json"]["schema"] + assert schema["properties"]["x"] == {"type": "number"} + assert schema["properties"]["y"] == {"type": "number"} + assert set(schema["required"]) == {"x", "y"} + assert body["required"] is True + + def test_optional_param_not_required(self): + app = mk_app([summarize]) + spec = export_openapi(app) + schema = spec["paths"]["/summarize"]["post"]["requestBody"]["content"][ + "application/json" + ]["schema"] + # ``count`` has a default → not required. + assert schema["required"] == ["label"] + + def test_response_schema_is_ref(self): + app = mk_app([make_point]) + spec = export_openapi(app) + response = spec["paths"]["/make_point"]["post"]["responses"]["200"] + schema = response["content"]["application/json"]["schema"] + assert schema == {"$ref": "#/components/schemas/Point"} + + def test_components_schemas_populated(self): + app = mk_app([make_point, summarize]) + spec = export_openapi(app) + schemas = spec["components"]["schemas"] + assert "Point" in schemas + assert "CorpusSummary" in schemas + + def test_get_route_uses_query_parameters(self): + # A convention-routed GET turns non-path args into query parameters, + # not a request body. + app = mk_app([get_item], use_conventions=True) + spec = export_openapi(app) + # Find the get_item operation regardless of the conventional path. + op = next( + methods["get"] + for methods in spec["paths"].values() + if "get" in methods + ) + assert "requestBody" not in op + names = {p["name"]: p for p in op.get("parameters", [])} + assert "item_id" in names + assert names["item_id"]["schema"] == {"type": "string"} + + def test_none_return_is_null_schema(self): + def drop(item_id: str) -> None: + """Delete something.""" + + app = mk_app([drop]) + spec = export_openapi(app) + schema = spec["paths"]["/drop"]["post"]["responses"]["200"]["content"][ + "application/json" + ]["schema"] + assert schema == {"type": "null"} + + def test_include_schemas_false_skips_request_body(self): + app = mk_app([make_point]) + spec = enhance_openapi_schema(app, include_schemas=False) + assert "requestBody" not in spec["paths"]["/make_point"]["post"] + + +class TestInstalledOnApp: + def test_mk_app_serves_enhanced_openapi_by_default(self): + app = mk_app([make_point]) + spec = app.openapi() # exercises the installed app.openapi override + body = spec["paths"]["/make_point"]["post"].get("requestBody") + assert body is not None + assert "Point" in spec.get("components", {}).get("schemas", {}) + + def test_enhanced_openapi_can_be_disabled(self): + app = mk_app([make_point], enhanced_openapi=False) + spec = app.openapi() + # FastAPI's plain schema has no request body for a Request-based endpoint. + assert "requestBody" not in spec["paths"]["/make_point"]["post"] + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) From 33981c021e8be8c55edbfced9ce109151212de21 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Thu, 21 May 2026 21:04:54 +0200 Subject: [PATCH 2/2] openapi: resolve forward-ref strings in type-hint conversion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_type_hints leaves the inner type of a self-referential annotation like list["TreeNode"] unresolved — a bare str on Python 3.10, a ForwardRef elsewhere. python_type_to_json_schema discarded both, so a self-referential field's items lost its $ref on 3.10. Resolve a forward-ref str/ForwardRef against the schema registry and recursion stack instead. Fixes the test_self_referential_dataclass_terminates failure on Python 3.10. --- qh/openapi.py | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/qh/openapi.py b/qh/openapi.py index e6f20bd..56bef83 100644 --- a/qh/openapi.py +++ b/qh/openapi.py @@ -41,6 +41,7 @@ Any, Callable, Dict, + ForwardRef, List, Literal, Optional, @@ -181,11 +182,23 @@ def python_type_to_json_schema( """ stack = _stack if _stack is not None else frozenset() - # Unknowns and unresolved string annotations → permissive empty schema. + # Unknowns → permissive empty schema. if type_hint is inspect.Parameter.empty or type_hint is Any: return {} - if isinstance(type_hint, str): - return {} + + # Forward references — a bare string or a typing.ForwardRef. These appear + # for self-referential types: ``get_type_hints`` leaves the inner type of + # ``list["TreeNode"]`` unresolved (a bare ``str`` on Python 3.10, a + # ``ForwardRef`` elsewhere). Resolve only against names already known to + # the schema registry or currently on the recursion stack; an otherwise + # unresolvable reference degrades to the permissive empty schema. + if isinstance(type_hint, (str, ForwardRef)): + name = ( + type_hint + if isinstance(type_hint, str) + else type_hint.__forward_arg__ + ) + return _ref(name) if (name in schemas or name in stack) else {} # Exact primitive match (covers ``NoneType`` too). if type_hint in _PRIMITIVE_SCHEMAS: