From a5896741f84d0c43691272a225eeefd1d88c0d1b Mon Sep 17 00:00:00 2001 From: Joakim Nordling Date: Thu, 23 Oct 2025 08:07:42 +0300 Subject: [PATCH 1/3] Improve handling of signatures for route handlers If a handler function had a modified function signature (like a header was removed from it), the generated OpenAPI spec could still end up having the parameter included due to how the handler was copied and the annotation for the request model was added. This changes how the handler function is copied and how the annotation for the request model is added so it works correctly in the case when the signature of the handler function has been modified. --- openapi_to_fastapi/tests/test_router.py | 42 ++++++++++++++++++++++++- openapi_to_fastapi/utils.py | 28 ++++++++++++++--- pyproject.toml | 2 +- 3 files changed, 65 insertions(+), 7 deletions(-) diff --git a/openapi_to_fastapi/tests/test_router.py b/openapi_to_fastapi/tests/test_router.py index 8744f61..f9e567d 100644 --- a/openapi_to_fastapi/tests/test_router.py +++ b/openapi_to_fastapi/tests/test_router.py @@ -1,4 +1,5 @@ -from typing import Any, Dict +import inspect +from typing import Any, Dict, Optional import pydantic import pytest @@ -596,3 +597,42 @@ def make_strict_request(json: Dict[str, Any]) -> Any: assert resp.status_code == expected_strict_code, resp.json() if resp.status_code != 200: assert json_snapshot == resp.json() + + +def test_modified_handler_signatures(app, client, specs_root): + + spec_router = SpecRouter(specs_root / "definitions") + + def handler_1(request, x_my_header: Optional[str] = Header(None)): + return {} + + def handler_2(request, x_my_header: Optional[str] = Header(None)): + return {} + + # Remove the header from handler_2 + sig = inspect.signature(handler_2) + params = sig.parameters + filtered_params = [ + param for param_name, param in params.items() if param_name != "x_my_header" + ] + handler_2.__signature__ = sig.replace(parameters=filtered_params) + + # Add handlers to router (non-decorator syntax) + spec_router.post("/TestValidation_v0.1")(handler_1) + spec_router.post("/TestValidation_v0.2")(handler_2) + + router = spec_router.to_fastapi_router() + app.include_router(router) + openapi_spec = app.openapi() + + route_spec_1 = openapi_spec["paths"]["/TestValidation_v0.1"] + route_spec_2 = openapi_spec["paths"]["/TestValidation_v0.2"] + + parameters_1 = route_spec_1["post"].get("parameters", {}) + parameters_2 = route_spec_2["post"].get("parameters", {}) + + headers_1 = {p.get("name") for p in parameters_1 if p.get("in") == "header"} + headers_2 = {p.get("name") for p in parameters_2 if p.get("in") == "header"} + + assert "x-my-header" in headers_1 + assert "x-my-header" not in headers_2 diff --git a/openapi_to_fastapi/utils.py b/openapi_to_fastapi/utils.py index af51b0d..93ef7a4 100644 --- a/openapi_to_fastapi/utils.py +++ b/openapi_to_fastapi/utils.py @@ -22,6 +22,11 @@ def copy_function(fn) -> Callable: ) g.__kwdefaults__ = deepcopy(fn.__kwdefaults__) g.__annotations__ = deepcopy(fn.__annotations__) + + # Signature is immutable, no need to copy/deepcopy + # Mypy doesn't know about __signature__: https://github.com/python/mypy/issues/12472 + g.__signature__ = inspect.signature(fn) # type: ignore[attr-defined] + return g @@ -32,9 +37,22 @@ def add_annotation_to_first_argument(fn: FunctionType, model: Type[pydantic.Base :param fn: Function to patch :param model: Type to add to the first argument """ - fn_spec = inspect.getfullargspec(fn) - if not len(fn_spec.args): + + sig = inspect.signature(fn) + params = sig.parameters + if not params: raise ValueError(f"Function {fn.__name__} has no arguments") - untyped_args = [a for a in fn_spec.args if a not in fn.__annotations__] - if untyped_args: - fn.__annotations__[untyped_args[0]] = model + + updated = False + updated_params = [] + for param_name, param in params.items(): + if not updated and param.annotation is inspect.Parameter.empty: + updated_params.append(param.replace(annotation=model)) + updated = True + else: + updated_params.append(param) + + # Mypy doesn't know about __signature__: https://github.com/python/mypy/issues/12472 + fn.__signature__ = sig.replace( # type: ignore[attr-defined] + parameters=updated_params + ) diff --git a/pyproject.toml b/pyproject.toml index 3b3f4f3..04ccc94 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "openapi-to-fastapi" -version = "0.20.0" +version = "0.21.0" description = "Create FastAPI routes from OpenAPI spec" authors = ["IOXIO Ltd"] license = "BSD-3-Clause" From df530a4c2e90bd0c99565a9c93469cb2fe2d4a01 Mon Sep 17 00:00:00 2001 From: Joakim Nordling Date: Thu, 23 Oct 2025 08:33:11 +0300 Subject: [PATCH 2/3] Improve tests by also adding a header --- openapi_to_fastapi/tests/test_router.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/openapi_to_fastapi/tests/test_router.py b/openapi_to_fastapi/tests/test_router.py index f9e567d..f70e181 100644 --- a/openapi_to_fastapi/tests/test_router.py +++ b/openapi_to_fastapi/tests/test_router.py @@ -609,13 +609,21 @@ def handler_1(request, x_my_header: Optional[str] = Header(None)): def handler_2(request, x_my_header: Optional[str] = Header(None)): return {} - # Remove the header from handler_2 + # Remove the header from handler_2 and add another header instead sig = inspect.signature(handler_2) params = sig.parameters - filtered_params = [ + modified_params = [ param for param_name, param in params.items() if param_name != "x_my_header" ] - handler_2.__signature__ = sig.replace(parameters=filtered_params) + modified_params.append( + inspect.Parameter( + "x_other_header", + kind=inspect.Parameter.KEYWORD_ONLY, + annotation=Optional[str], + default=Header(None), + ) + ) + handler_2.__signature__ = sig.replace(parameters=modified_params) # Add handlers to router (non-decorator syntax) spec_router.post("/TestValidation_v0.1")(handler_1) @@ -634,5 +642,5 @@ def handler_2(request, x_my_header: Optional[str] = Header(None)): headers_1 = {p.get("name") for p in parameters_1 if p.get("in") == "header"} headers_2 = {p.get("name") for p in parameters_2 if p.get("in") == "header"} - assert "x-my-header" in headers_1 - assert "x-my-header" not in headers_2 + assert headers_1 == {"x-my-header"} + assert headers_2 == {"x-other-header"} From ec99118d37b71a9c7b611c3447feeef68d4dd228 Mon Sep 17 00:00:00 2001 From: Joakim Nordling Date: Thu, 23 Oct 2025 10:35:25 +0300 Subject: [PATCH 3/3] Revert "Improve tests by also adding a header" This reverts commit df530a4c2e90bd0c99565a9c93469cb2fe2d4a01. If you would call that route, you would get extra arguments that the function can not handle. I.e. if you use the approach of changing the signature, you should rather just remove optional parameters than trying to add more. --- openapi_to_fastapi/tests/test_router.py | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/openapi_to_fastapi/tests/test_router.py b/openapi_to_fastapi/tests/test_router.py index f70e181..f9e567d 100644 --- a/openapi_to_fastapi/tests/test_router.py +++ b/openapi_to_fastapi/tests/test_router.py @@ -609,21 +609,13 @@ def handler_1(request, x_my_header: Optional[str] = Header(None)): def handler_2(request, x_my_header: Optional[str] = Header(None)): return {} - # Remove the header from handler_2 and add another header instead + # Remove the header from handler_2 sig = inspect.signature(handler_2) params = sig.parameters - modified_params = [ + filtered_params = [ param for param_name, param in params.items() if param_name != "x_my_header" ] - modified_params.append( - inspect.Parameter( - "x_other_header", - kind=inspect.Parameter.KEYWORD_ONLY, - annotation=Optional[str], - default=Header(None), - ) - ) - handler_2.__signature__ = sig.replace(parameters=modified_params) + handler_2.__signature__ = sig.replace(parameters=filtered_params) # Add handlers to router (non-decorator syntax) spec_router.post("/TestValidation_v0.1")(handler_1) @@ -642,5 +634,5 @@ def handler_2(request, x_my_header: Optional[str] = Header(None)): headers_1 = {p.get("name") for p in parameters_1 if p.get("in") == "header"} headers_2 = {p.get("name") for p in parameters_2 if p.get("in") == "header"} - assert headers_1 == {"x-my-header"} - assert headers_2 == {"x-other-header"} + assert "x-my-header" in headers_1 + assert "x-my-header" not in headers_2