diff --git a/spp_api_v2/README.rst b/spp_api_v2/README.rst
index 799bfd999..6fc38c2aa 100644
--- a/spp_api_v2/README.rst
+++ b/spp_api_v2/README.rst
@@ -147,6 +147,25 @@ Dependencies
Changelog
=========
+19.0.2.1.0
+~~~~~~~~~~
+
+- Add OpenAPI polymorphic schema utilities
+ (``utils/openapi_polymorphic.py``): ``polymorphic_body()`` for
+ declaring dict-typed fields that accept one of several Pydantic
+ models, plus an app-level OpenAPI hook that injects the corresponding
+ ``anyOf`` schemas into the generated document
+- Auth middleware: replace the plain ``HTTPBearer`` scheme with an
+ OAuth2 client-credentials security scheme so the OpenAPI document
+ advertises the token endpoint and consumers (Swagger UI, QGIS, etc.)
+ can discover how to authenticate; strip the ``Bearer`` prefix from the
+ raw Authorization header value
+- Bundle schemas: document ``BundleEntry.resource`` as a polymorphic
+ Individual/Group body instead of an opaque dict, so bundle payloads
+ are fully described in the OpenAPI document
+- Add OpenAPI contract tests covering bundle schema rendering, the
+ polymorphic utilities, and the overall OpenAPI document contract
+
19.0.2.0.1
~~~~~~~~~~
diff --git a/spp_api_v2/__manifest__.py b/spp_api_v2/__manifest__.py
index cfbe15748..b1d1caa1a 100644
--- a/spp_api_v2/__manifest__.py
+++ b/spp_api_v2/__manifest__.py
@@ -1,7 +1,7 @@
{
"name": "OpenSPP API V2",
"category": "OpenSPP/Integration",
- "version": "19.0.2.0.1",
+ "version": "19.0.2.1.0",
"sequence": 1,
"author": "OpenSPP.org",
"website": "https://github.com/OpenSPP/OpenSPP2",
diff --git a/spp_api_v2/middleware/auth.py b/spp_api_v2/middleware/auth.py
index c914ba885..4e0dcbe84 100644
--- a/spp_api_v2/middleware/auth.py
+++ b/spp_api_v2/middleware/auth.py
@@ -10,13 +10,24 @@
from odoo.addons.fastapi.dependencies import odoo_env
from fastapi import Depends, HTTPException, status
-from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from fastapi.security import OAuth2
_logger = logging.getLogger(__name__)
-# HTTP Bearer scheme for extracting token from Authorization header
-# auto_error=False allows us to handle authentication errors with proper status codes
-security = HTTPBearer(auto_error=False)
+# OAuth2 Client Credentials scheme.
+# This produces an "oauth2" entry in the OpenAPI securitySchemes with the
+# clientCredentials flow pointing at our token endpoint, so API consumers
+# (Swagger UI, QGIS, etc.) can discover how to authenticate.
+# auto_error=False allows us to handle authentication errors with proper status codes.
+security = OAuth2(
+ flows={
+ "clientCredentials": {
+ "tokenUrl": "oauth/token",
+ "scopes": {},
+ },
+ },
+ auto_error=False,
+)
# Cache for JWT secret validation results, keyed by hash of the secret.
# Avoids recomputing Shannon entropy on every API request.
@@ -24,7 +35,7 @@
def get_authenticated_client(
- credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(security)],
+ token: Annotated[str | None, Depends(security)],
env: Annotated[Environment, Depends(odoo_env)],
):
"""
@@ -39,14 +50,17 @@ def get_authenticated_client(
Raises:
HTTPException: If token is invalid, expired, or client not found
"""
- if not credentials:
+ if not token:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Missing Authorization header",
headers={"WWW-Authenticate": "Bearer"},
)
- token = credentials.credentials
+ # OAuth2 dependency returns the full Authorization header value
+ # (e.g. "Bearer eyJ..."). Strip the scheme prefix to get the raw JWT.
+ if token.lower().startswith("bearer "):
+ token = token[7:].strip()
try:
# Decode and validate JWT
@@ -91,7 +105,7 @@ def get_authenticated_client(
def get_current_client(
- credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
+ token: Annotated[str | None, Depends(security)],
env: Annotated[Environment, Depends(odoo_env)],
) -> dict:
"""
@@ -103,7 +117,7 @@ def get_current_client(
Returns:
dict: {"env": Environment, "client": spp.api.client record}
"""
- client = get_authenticated_client(credentials, env)
+ client = get_authenticated_client(token, env)
return {"env": env, "client": client}
diff --git a/spp_api_v2/models/fastapi_endpoint_registry.py b/spp_api_v2/models/fastapi_endpoint_registry.py
index dd151c238..c8b1258b4 100644
--- a/spp_api_v2/models/fastapi_endpoint_registry.py
+++ b/spp_api_v2/models/fastapi_endpoint_registry.py
@@ -5,6 +5,8 @@
from fastapi import APIRouter, FastAPI
+from ..utils.openapi_polymorphic import install_polymorphic_openapi_hook
+
_logger = logging.getLogger(__name__)
@@ -79,6 +81,8 @@ def _get_app(self) -> FastAPI:
from ..middleware.version import VersionMiddleware
app.add_middleware(VersionMiddleware)
+ # Install OpenAPI hook so polymorphic_body() schemas are injected
+ install_polymorphic_openapi_hook(app)
# V2 API uses public endpoint with JWT authentication in middleware
# No default authentication required at app level
return app
diff --git a/spp_api_v2/readme/HISTORY.md b/spp_api_v2/readme/HISTORY.md
index 375bed600..ce73d6afe 100644
--- a/spp_api_v2/readme/HISTORY.md
+++ b/spp_api_v2/readme/HISTORY.md
@@ -1,3 +1,10 @@
+### 19.0.2.1.0
+
+- Add OpenAPI polymorphic schema utilities (`utils/openapi_polymorphic.py`): `polymorphic_body()` for declaring dict-typed fields that accept one of several Pydantic models, plus an app-level OpenAPI hook that injects the corresponding `anyOf` schemas into the generated document
+- Auth middleware: replace the plain `HTTPBearer` scheme with an OAuth2 client-credentials security scheme so the OpenAPI document advertises the token endpoint and consumers (Swagger UI, QGIS, etc.) can discover how to authenticate; strip the `Bearer` prefix from the raw Authorization header value
+- Bundle schemas: document `BundleEntry.resource` as a polymorphic Individual/Group body instead of an opaque dict, so bundle payloads are fully described in the OpenAPI document
+- Add OpenAPI contract tests covering bundle schema rendering, the polymorphic utilities, and the overall OpenAPI document contract
+
### 19.0.2.0.1
- Fix `SerializationFailure` race when multiple Odoo workers rebuild their routing map simultaneously (e.g. after `-u all`) and all try to sync the same `fastapi.endpoint` rows
diff --git a/spp_api_v2/schemas/__init__.py b/spp_api_v2/schemas/__init__.py
index c85d70a75..e9f4137af 100644
--- a/spp_api_v2/schemas/__init__.py
+++ b/spp_api_v2/schemas/__init__.py
@@ -1,10 +1,10 @@
from . import api_metadata
from . import base
from . import bulk
-from . import bundle
from . import filter as filter_schema
from . import group
from . import individual
+from . import bundle # depends on individual + group for polymorphic_body refs
from . import membership
from . import operation_outcome
from . import patch
diff --git a/spp_api_v2/schemas/bundle.py b/spp_api_v2/schemas/bundle.py
index 511c031aa..ff2629c1a 100644
--- a/spp_api_v2/schemas/bundle.py
+++ b/spp_api_v2/schemas/bundle.py
@@ -5,6 +5,11 @@
from pydantic import BaseModel, ConfigDict, Field
+from odoo.addons.spp_api_v2.utils.openapi_polymorphic import polymorphic_body
+
+from .group import Group
+from .individual import Individual
+
class BundleLink(BaseModel):
"""Link in a bundle (for pagination)"""
@@ -50,7 +55,12 @@ class BundleEntry(BaseModel):
)
request: BundleRequest | None = None
response: BundleResponse | None = None
- resource: dict[str, Any] | None = None
+ resource: dict[str, Any] | None = polymorphic_body(
+ Individual,
+ Group,
+ default=None,
+ description="FHIR-style resource. Must match the type indicated by request.url.",
+ )
search: BundleSearch | None = None
diff --git a/spp_api_v2/static/description/index.html b/spp_api_v2/static/description/index.html
index ca6be49f3..562d9124c 100644
--- a/spp_api_v2/static/description/index.html
+++ b/spp_api_v2/static/description/index.html
@@ -517,18 +517,39 @@ Dependencies
-
+
+
+- Add OpenAPI polymorphic schema utilities
+(utils/openapi_polymorphic.py): polymorphic_body() for
+declaring dict-typed fields that accept one of several Pydantic
+models, plus an app-level OpenAPI hook that injects the corresponding
+anyOf schemas into the generated document
+- Auth middleware: replace the plain HTTPBearer scheme with an
+OAuth2 client-credentials security scheme so the OpenAPI document
+advertises the token endpoint and consumers (Swagger UI, QGIS, etc.)
+can discover how to authenticate; strip the Bearer prefix from the
+raw Authorization header value
+- Bundle schemas: document BundleEntry.resource as a polymorphic
+Individual/Group body instead of an opaque dict, so bundle payloads
+are fully described in the OpenAPI document
+- Add OpenAPI contract tests covering bundle schema rendering, the
+polymorphic utilities, and the overall OpenAPI document contract
+
+
+
+
- Fix SerializationFailure race when multiple Odoo workers rebuild
their routing map simultaneously (e.g. after -u all) and all try
@@ -543,15 +564,15 @@
regressions are diagnosable without raising the global log level
-
-
+
+
- Initial migration to OpenSPP2
-
+
Bugs are tracked on GitHub Issues.
In case of trouble, please check there if your issue has already been reported.
If you spotted it first, help us to smash it by providing a detailed and welcomed
@@ -559,7 +580,7 @@
Do not contact contributors directly about support or help with technical issues.
diff --git a/spp_api_v2/tests/__init__.py b/spp_api_v2/tests/__init__.py
index 806558639..0a9e53c0d 100644
--- a/spp_api_v2/tests/__init__.py
+++ b/spp_api_v2/tests/__init__.py
@@ -12,6 +12,7 @@
from . import test_audit_log_performance
from . import test_batch_api
from . import test_bulk_api
+from . import test_bundle_openapi
from . import test_bundle_service
from . import test_consent
from . import test_consent_history
@@ -28,6 +29,8 @@
from . import test_jwt_secret_validation
from . import test_metadata
from . import test_oauth
+from . import test_openapi_contract
+from . import test_openapi_polymorphic
from . import test_organization_type_security
from . import test_pagination
from . import test_patch_api
diff --git a/spp_api_v2/tests/test_bundle_openapi.py b/spp_api_v2/tests/test_bundle_openapi.py
new file mode 100644
index 000000000..e4e433e22
--- /dev/null
+++ b/spp_api_v2/tests/test_bundle_openapi.py
@@ -0,0 +1,47 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""OpenAPI-shape tests for the Bundle schema.
+
+Asserts that `BundleEntry.resource` documents its accepted resource types
+(Individual, Group) via `oneOf` of $refs instead of a bare `dict | None`.
+"""
+
+from odoo.tests.common import HttpCase, tagged
+
+
+@tagged("post_install", "-at_install")
+class TestBundleEntryOpenAPI(HttpCase):
+ """Bundle schema renders polymorphic resource documentation."""
+
+ def test_bundle_entry_resource_documented_as_oneof(self):
+ """BundleEntry.resource should document oneOf of supported FHIR types.
+
+ Bundle service only supports Individual and Group (see
+ spp_api_v2/services/bundle_service.py:299, 324, 350); anything else
+ is rejected at runtime, so oneOf must list exactly those.
+ """
+ response = self.url_open("/api/v2/spp/openapi.json")
+ self.assertEqual(response.status_code, 200, response.text)
+ schema = response.json()
+ components = schema["components"]["schemas"]
+
+ self.assertIn("BundleEntry", components)
+ resource_schema = components["BundleEntry"]["properties"]["resource"]
+
+ # Spike-confirmed shape: for `dict | None = polymorphic_body(...)`,
+ # Pydantic emits `anyOf: [{type: object}, {type: null}]` and our hook
+ # attaches `oneOf` at the SAME top level (siblings, not nested).
+ self.assertIn("oneOf", resource_schema, f"no oneOf at top level: {resource_schema}")
+ refs = [item.get("$ref") for item in resource_schema["oneOf"]]
+ self.assertIn("#/components/schemas/Individual", refs)
+ self.assertIn("#/components/schemas/Group", refs)
+
+ # And the nullable shape comes from anyOf alongside.
+ self.assertIn(
+ {"type": "null"},
+ resource_schema.get("anyOf", []),
+ f"missing nullable anyOf branch: {resource_schema}",
+ )
+
+ # Both referenced models must actually be present in components.
+ self.assertIn("Individual", components)
+ self.assertIn("Group", components)
diff --git a/spp_api_v2/tests/test_openapi_contract.py b/spp_api_v2/tests/test_openapi_contract.py
new file mode 100644
index 000000000..aa8f3a316
--- /dev/null
+++ b/spp_api_v2/tests/test_openapi_contract.py
@@ -0,0 +1,45 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Contract test: every $ref in the OpenAPI schema must resolve.
+
+Catches the failure mode where polymorphic_body declares a oneOf of $refs
+but the referenced model isn't registered or the OpenAPI hook isn't
+installed.
+"""
+
+from odoo.tests.common import HttpCase, tagged
+
+
+@tagged("post_install", "-at_install")
+class TestOpenAPIContract(HttpCase):
+ """Walk the live OpenAPI schema; assert every $ref resolves."""
+
+ def test_all_refs_resolve(self):
+ response = self.url_open("/api/v2/spp/openapi.json")
+ self.assertEqual(response.status_code, 200, response.text)
+ schema = response.json()
+ components = schema.get("components", {}).get("schemas", {})
+
+ unresolved = []
+
+ def walk(node, path):
+ if isinstance(node, dict):
+ ref = node.get("$ref")
+ if isinstance(ref, str) and ref.startswith("#/components/schemas/"):
+ name = ref.rsplit("/", 1)[-1]
+ if name not in components:
+ unresolved.append((path, ref))
+ for k, v in node.items():
+ walk(v, f"{path}.{k}")
+ elif isinstance(node, list):
+ for i, v in enumerate(node):
+ walk(v, f"{path}[{i}]")
+
+ walk(schema, "$")
+
+ self.assertEqual(
+ unresolved,
+ [],
+ "Unresolved $refs in OpenAPI schema. Either install_polymorphic_openapi_hook "
+ "is not wired, or a polymorphic_body() references a model not in components/schemas.\n"
+ f"Found: {unresolved[:5]}",
+ )
diff --git a/spp_api_v2/tests/test_openapi_polymorphic.py b/spp_api_v2/tests/test_openapi_polymorphic.py
new file mode 100644
index 000000000..f368bcee7
--- /dev/null
+++ b/spp_api_v2/tests/test_openapi_polymorphic.py
@@ -0,0 +1,182 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Unit tests for the polymorphic_body helper and OpenAPI hook.
+
+These tests do not require Odoo; they construct minimal FastAPI apps
+and assert against the generated OpenAPI schema.
+"""
+
+from typing import Annotated
+
+from fastapi.testclient import TestClient
+from pydantic import BaseModel, Field
+
+from odoo.tests.common import TransactionCase
+
+from fastapi import Body, FastAPI
+
+from ..utils.openapi_polymorphic import (
+ install_polymorphic_openapi_hook,
+ polymorphic_body,
+ reset_polymorphic_registry,
+)
+
+
+class SimpleA(BaseModel):
+ a_field: str = Field(..., description="A field")
+
+
+class SimpleB(BaseModel):
+ b_field: int
+
+
+def _make_app():
+ reset_polymorphic_registry()
+ app = FastAPI(title="Test", version="0.0.1")
+
+ class Body_(BaseModel):
+ payload: dict = polymorphic_body(SimpleA, SimpleB, description="Payload")
+
+ @app.post("/echo")
+ def echo(body: Annotated[Body_, Body(...)]) -> dict:
+ return {"type": type(body.payload).__name__, "value": body.payload}
+
+ install_polymorphic_openapi_hook(app)
+ return app
+
+
+class TestPolymorphicBody(TransactionCase):
+ """Unit tests for polymorphic_body helper."""
+
+ def test_runtime_payload_is_dict(self):
+ """Bodies matching SimpleA stay as dict, not parsed into a model."""
+ client = TestClient(_make_app())
+ r = client.post("/echo", json={"payload": {"a_field": "x"}})
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.json(), {"type": "dict", "value": {"a_field": "x"}})
+
+ def test_runtime_arbitrary_dict_accepted(self):
+ """Bodies matching neither schema still pass (no validation regression)."""
+ client = TestClient(_make_app())
+ r = client.post("/echo", json={"payload": {"random": "stuff"}})
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.json()["value"], {"random": "stuff"})
+
+ def test_openapi_field_has_oneof_refs(self):
+ """The body field renders as oneOf of $refs."""
+ schema = _make_app().openapi()
+ body_schema = schema["components"]["schemas"]["Body_"]
+ payload = body_schema["properties"]["payload"]
+ self.assertEqual(
+ payload["oneOf"],
+ [
+ {"$ref": "#/components/schemas/SimpleA"},
+ {"$ref": "#/components/schemas/SimpleB"},
+ ],
+ )
+
+ def test_openapi_referenced_models_present(self):
+ """Both typed models appear in components/schemas."""
+ schema = _make_app().openapi()
+ components = schema["components"]["schemas"]
+ self.assertIn("SimpleA", components)
+ self.assertIn("SimpleB", components)
+ self.assertEqual(components["SimpleA"]["properties"]["a_field"]["description"], "A field")
+
+ def test_openapi_unreferenced_models_not_injected(self):
+ """Models registered but not referenced by any route are not added.
+
+ Scoping check: registering SimpleA via polymorphic_body in one app
+ must not cause it to leak into a different app that doesn't use it.
+ """
+ reset_polymorphic_registry()
+
+ class WithBody(BaseModel):
+ payload: dict = polymorphic_body(SimpleA, description="x")
+
+ # First app uses the helper.
+ app1 = FastAPI()
+
+ @app1.post("/x")
+ def _x(body: Annotated[WithBody, Body(...)]) -> dict:
+ return {}
+
+ install_polymorphic_openapi_hook(app1)
+ self.assertIn("SimpleA", app1.openapi()["components"]["schemas"])
+
+ # Second app does NOT use the helper but shares the registry.
+ app2 = FastAPI()
+
+ @app2.get("/y")
+ def _y() -> dict:
+ return {}
+
+ install_polymorphic_openapi_hook(app2)
+ self.assertNotIn(
+ "SimpleA",
+ app2.openapi().get("components", {}).get("schemas", {}),
+ )
+
+ def test_openapi_nested_model_refs_resolve(self):
+ """Nested Pydantic types referenced by registered models also get injected."""
+ reset_polymorphic_registry()
+
+ class Inner(BaseModel):
+ n: int
+
+ class Outer(BaseModel):
+ inner: Inner
+
+ class Req(BaseModel):
+ payload: dict = polymorphic_body(Outer)
+
+ app = FastAPI()
+
+ @app.post("/z")
+ def _z(body: Annotated[Req, Body(...)]) -> dict:
+ return {}
+
+ install_polymorphic_openapi_hook(app)
+ components = app.openapi()["components"]["schemas"]
+ self.assertIn("Outer", components)
+ self.assertIn("Inner", components)
+ self.assertEqual(
+ components["Outer"]["properties"]["inner"]["$ref"],
+ "#/components/schemas/Inner",
+ )
+
+ def test_polymorphic_body_supports_optional_default(self):
+ """Optional fields render correctly with `default=None`.
+
+ For a `dict | None = polymorphic_body(..., default=None)` field, Pydantic
+ emits `anyOf: [{type: object}, {type: null}]` and our hook attaches
+ `oneOf: []` at the same level. The two siblings are combined by
+ JSON-Schema AND semantics; this is the documented (and Swagger-renders-
+ correctly) trade-off, see the plan's "Open Questions" section.
+ """
+ reset_polymorphic_registry()
+
+ class Req(BaseModel):
+ payload: dict | None = polymorphic_body(SimpleA, default=None)
+
+ app = FastAPI()
+
+ @app.post("/x")
+ def _x(body: Annotated[Req, Body(...)]) -> dict:
+ return {"none": body.payload is None}
+
+ install_polymorphic_openapi_hook(app)
+
+ schema = app.openapi()
+ payload = schema["components"]["schemas"]["Req"]["properties"]["payload"]
+ self.assertIn("oneOf", payload)
+ self.assertEqual(payload["oneOf"], [{"$ref": "#/components/schemas/SimpleA"}])
+ # Pydantic adds the nullable shape as `anyOf`.
+ self.assertIn({"type": "null"}, payload.get("anyOf", []))
+
+ # And the model still resolves at runtime.
+ self.assertIn("SimpleA", schema["components"]["schemas"])
+
+ client = TestClient(app)
+ r = client.post("/x", json={})
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.json(), {"none": True})
diff --git a/spp_api_v2/utils/openapi_polymorphic.py b/spp_api_v2/utils/openapi_polymorphic.py
new file mode 100644
index 000000000..557954566
--- /dev/null
+++ b/spp_api_v2/utils/openapi_polymorphic.py
@@ -0,0 +1,154 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Helpers for documenting polymorphic dict-typed request bodies.
+
+A polymorphic body is a request field whose Python type stays `dict` (so
+downstream code can keep calling `.get(...)`) but whose OpenAPI schema is
+documented as a `oneOf` of typed Pydantic models.
+
+Usage in a schema:
+
+ from ..utils.openapi_polymorphic import polymorphic_body
+
+ class ExecuteRequest(BaseModel):
+ inputs: dict = polymorphic_body(
+ SpatialStatisticsInputs,
+ ProximityStatisticsInputs,
+ description="Process input values; structure depends on process_id.",
+ )
+
+Wiring the hook (once per FastAPI app):
+
+ from .utils.openapi_polymorphic import install_polymorphic_openapi_hook
+ install_polymorphic_openapi_hook(app)
+
+Why a hook is needed: models passed to `polymorphic_body` are referenced only
+via `$ref` strings in `json_schema_extra`. FastAPI's OpenAPI generator does not
+discover them from endpoint signatures, so the hook injects them into
+`components/schemas` after generation.
+"""
+
+import logging
+from typing import Any
+
+from fastapi.openapi.utils import get_openapi
+from pydantic import BaseModel, Field
+from pydantic.json_schema import models_json_schema
+
+from fastapi import FastAPI
+
+_logger = logging.getLogger(__name__)
+
+# Module-level registry. Shared across all FastAPI apps in the process; the
+# hook scopes injection to models actually referenced by each app's routes.
+_REGISTRY: list[type[BaseModel]] = []
+
+
+def polymorphic_body(
+ *models: type[BaseModel],
+ description: str = "",
+ default: Any = ...,
+) -> Any:
+ """Return a Pydantic Field documenting a dict body as `oneOf` of models.
+
+ The runtime type stays `dict` (or `dict | None` if `default=None` and the
+ annotation is `dict | None`). Downstream validation and access patterns
+ are unaffected. Models are registered for OpenAPI injection by
+ `install_polymorphic_openapi_hook`.
+
+ For optional fields (e.g. `BundleEntry.resource`), pass `default=None`
+ alongside a `dict | None` annotation.
+ """
+ if not models:
+ raise ValueError("polymorphic_body requires at least one model")
+ for m in models:
+ if not (isinstance(m, type) and issubclass(m, BaseModel)):
+ raise TypeError(f"polymorphic_body expects BaseModel subclasses, got {m!r}")
+ if m not in _REGISTRY:
+ _REGISTRY.append(m)
+ return Field(
+ default,
+ description=description,
+ json_schema_extra={
+ "oneOf": [{"$ref": f"#/components/schemas/{m.__name__}"} for m in models],
+ },
+ )
+
+
+def register_polymorphic_models(*models: type[BaseModel]) -> None:
+ """Register models for OpenAPI injection without attaching them to a Field.
+
+ Useful when a polymorphic schema is built outside `polymorphic_body`
+ (e.g., for response unions). Idempotent.
+ """
+ for m in models:
+ if m not in _REGISTRY:
+ _REGISTRY.append(m)
+
+
+def reset_polymorphic_registry() -> None:
+ """Clear the registry. Test-only."""
+ _REGISTRY.clear()
+
+
+def _collect_refs(node: Any, refs: set[str] | None = None) -> set[str]:
+ """Recursively collect every $ref string from a JSON-Schema-shaped object."""
+ if refs is None:
+ refs = set()
+ if isinstance(node, dict):
+ ref = node.get("$ref")
+ if isinstance(ref, str):
+ refs.add(ref)
+ for v in node.values():
+ _collect_refs(v, refs)
+ elif isinstance(node, list):
+ for v in node:
+ _collect_refs(v, refs)
+ return refs
+
+
+def install_polymorphic_openapi_hook(app: FastAPI) -> None:
+ """Wrap `app.openapi` so registered models referenced by this app's routes
+ are injected into `components/schemas`.
+
+ Models registered globally but not referenced by any route in `app` are
+ NOT injected, so cross-app pollution is avoided.
+ """
+
+ def custom_openapi():
+ if app.openapi_schema:
+ return app.openapi_schema
+ schema = get_openapi(
+ title=app.title,
+ version=app.version,
+ description=app.description,
+ routes=app.routes,
+ )
+ components = schema.setdefault("components", {}).setdefault("schemas", {})
+
+ all_refs = _collect_refs(schema)
+ missing = {ref.rsplit("/", 1)[-1] for ref in all_refs if ref.startswith("#/components/schemas/")} - set(
+ components.keys()
+ )
+
+ wanted = [m for m in _REGISTRY if m.__name__ in missing]
+ if wanted:
+ _, defs = models_json_schema(
+ [(m, "validation") for m in wanted],
+ ref_template="#/components/schemas/{model}",
+ )
+ generated = defs.get("$defs", {})
+ if not generated:
+ _logger.warning(
+ "polymorphic OpenAPI hook found %d unresolved $refs (%s) "
+ "but models_json_schema produced no $defs; the schema may "
+ "still have dangling refs. Pydantic shape may have changed.",
+ len(wanted),
+ ", ".join(m.__name__ for m in wanted),
+ )
+ for name, model_schema in generated.items():
+ components.setdefault(name, model_schema)
+
+ app.openapi_schema = schema
+ return schema
+
+ app.openapi = custom_openapi