127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
"""
|
|
# Summary
|
|
|
|
OpenAPI-driven route binding for FastAPI.
|
|
|
|
This module is responsible for translating an OpenAPI 3.x specification
|
|
into concrete FastAPI routes. It enforces a strict one-to-one mapping
|
|
between OpenAPI operations and Python handler functions using `operationId`.
|
|
|
|
Notes:
|
|
**Core Responsibility:**
|
|
|
|
- Read path + method definitions from an OpenAPI specification.
|
|
- Resolve each `operationId` to a Python callable.
|
|
- Register routes with FastAPI using `APIRoute`.
|
|
- Fail fast when contract violations are detected.
|
|
|
|
**Design Constraints:**
|
|
|
|
- All routes MUST be declared in the OpenAPI specification.
|
|
- All OpenAPI operations MUST define an `operationId`.
|
|
- Every `operationId` MUST resolve to a handler function.
|
|
- Handlers are plain Python callables (no decorators required).
|
|
- No implicit route creation or inference is allowed.
|
|
|
|
**Constraints:**
|
|
|
|
- This module intentionally does NOT:
|
|
- Perform request or response validation.
|
|
- Generate Pydantic models.
|
|
- Modify FastAPI dependency injection.
|
|
- Interpret OpenAPI semantics beyond routing metadata.
|
|
"""
|
|
|
|
from types import ModuleType
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.routing import APIRoute
|
|
|
|
from .errors import MissingOperationHandler
|
|
|
|
|
|
def bind_routes(
|
|
app: FastAPI,
|
|
spec: dict[str, Any],
|
|
routes_module: ModuleType,
|
|
security_deps: dict[str, list[Any]] | None = None,
|
|
) -> None:
|
|
"""
|
|
Bind OpenAPI operations to FastAPI routes.
|
|
|
|
Args:
|
|
app (FastAPI):
|
|
The FastAPI application instance to which routes will be added.
|
|
spec (dict[str, Any]):
|
|
Parsed OpenAPI 3.x specification dictionary.
|
|
routes_module (ModuleType):
|
|
Python module containing handler functions. Each handler's name MUST
|
|
exactly match an OpenAPI `operationId`.
|
|
security_deps (dict[str, list[Any]] | None):
|
|
Optional mapping of ``METHOD:/path`` → ``list[Depends(...)]``
|
|
generated from the spec's ``securitySchemes`` and per-operation
|
|
``security`` fields.
|
|
|
|
Raises:
|
|
MissingOperationHandler:
|
|
If an ``operationId`` is missing from the spec or if no corresponding
|
|
handler function exists in the routes module.
|
|
|
|
Notes:
|
|
**Responsibilities:**
|
|
|
|
- Iterates through the OpenAPI specification paths and methods.
|
|
- Resolves each ``operationId`` to a handler function, and registers
|
|
a corresponding ``APIRoute`` on the FastAPI application.
|
|
- Injects FastAPI ``Depends()`` for each security requirement found
|
|
on the operation or inherited from the top-level ``security`` field.
|
|
|
|
**Guarantees:**
|
|
|
|
- Route registration is deterministic and spec-driven. No route
|
|
decorators are required or supported. Handler resolution errors
|
|
surface at application startup.
|
|
"""
|
|
|
|
paths = spec.get("paths", {})
|
|
security_deps = security_deps or {}
|
|
|
|
http_methods = {"get", "put", "post", "delete", "options", "head", "patch", "trace"}
|
|
|
|
for path, methods in paths.items():
|
|
for http_method, operation in methods.items():
|
|
if http_method.lower() not in http_methods or not isinstance(
|
|
operation, dict
|
|
):
|
|
continue
|
|
|
|
operation_id = operation.get("operationId")
|
|
if not operation_id:
|
|
raise MissingOperationHandler(
|
|
path=path,
|
|
method=http_method,
|
|
)
|
|
|
|
try:
|
|
endpoint = getattr(routes_module, operation_id)
|
|
except AttributeError:
|
|
raise MissingOperationHandler(
|
|
path=path,
|
|
method=http_method,
|
|
operation_id=operation_id,
|
|
) from None
|
|
|
|
key = f"{http_method.upper()}:{path}"
|
|
deps = security_deps.get(key, [])
|
|
|
|
route = APIRoute(
|
|
path=path,
|
|
endpoint=endpoint,
|
|
methods=[http_method.upper()],
|
|
dependencies=deps,
|
|
name=operation_id,
|
|
)
|
|
|
|
app.router.routes.append(route)
|