fix(router): support PEP 604 unions (pipe operator)

.pre-commit-config.yaml

@@ -52,7 +52,11 @@ repos:
     rev: v2.12.2
     hooks:
       - id: pylint
-        exclude: ^.*(docs/).*$
+        exclude: |
+            (?x)^(
+                .*(docs/).*|
+                .*py3_10.py
+            )$
         # disabled import-error as may be run out of environment with deps
         args: ["--disable=import-error"]
         additional_dependencies: ['toml', 'pyproject']

flask_ninja/models.py

@@ -152,7 +152,7 @@ class MediaType(BaseModel):
 class ParameterBase(BaseModel):
     description: Optional[str] = None
     required: Optional[bool] = None
-    deprecated: Optional[bool] = None
+    deprecated: Union[Any, str, bool, None] = None
     # Serialization rules for simple scenarios
     style: Optional[str] = None
     explode: Optional[bool] = None

flask_ninja/operation.py

@@ -10,7 +10,7 @@
 from pydantic import BaseModel, ConfigDict, ValidationError
 
 from .constants import NOT_SET, ApiConfigError, ParamType
-from .model_field import FieldMapping, ModelField, Undefined
+from .model_field import FieldMapping, ModelField, Undefined, UnionType
 from .models import MediaType
 from .models import Operation as OAPIOperation
 from .models import (
@@ -158,7 +158,7 @@ def _sanitize_responses(
 
         # Check if for each returned type there is implicitly or explicitly defined response model and code
         if func_return_type:
-            if get_origin(func_return_type) == Union:
+            if get_origin(func_return_type) in [Union, UnionType]:
                 for ret_type in func_return_type.__args__:
                     if not any(resp.type_ != ret_type for resp in responses.values()):
                         raise ApiConfigError(
@@ -217,14 +217,18 @@ def get_callback_schema(
             parameters.append(parameter)
 
         schema = OAPIOperation(
-            requestBody=RequestBody(
-                content={
-                    "application/json": MediaType(schema_=request_body)  # type:ignore
-                },
-                required=True,
-            )
-            if request_body
-            else None,
+            requestBody=(
+                RequestBody(
+                    content={
+                        "application/json": MediaType(
+                            schema_=request_body  # type:ignore
+                        )
+                    },
+                    required=True,
+                )
+                if request_body
+                else None
+            ),
             parameters=parameters or None,
             responses={
                 str(code): Response(description=description)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "flask-ninja"
-version = "1.3.0"
+version = "1.3.1"
 description = "Flask Ninja is a web framework for building APIs with Flask and Python 3.9+ type hints."
 readme = "README.md"
 authors = ["Michal Korbela <[email protected]>"]

tests/unit/snapshots/test_api/test_get_schema/api_schema

@@ -1 +1 @@
-{"openapi": "3.1.0", "info": {"title": "", "description": "", "version": "1.0.0"}, "paths": {"/some_endpoint/{param}": {"get": {"summary": "", "description": "", "parameters": [{"required": true, "schema": {"type": "integer"}, "name": "param", "in": "path"}], "requestBody": {"description": "", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Server"}}}, "required": true}, "responses": {"200": {"description": "", "content": {"application/json": {"schema": {"type": "integer"}}}}}, "security": [{"bearerTokenAuth": []}]}}}, "components": {"schemas": {"Server": {"additionalProperties": true, "properties": {"url": {"anyOf": [{"format": "uri", "minLength": 1, "type": "string"}, {"type": "string"}], "title": "Url"}, "description": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": null, "title": "Description"}, "variables": {"anyOf": [{"additionalProperties": {"$ref": "#/components/schemas/ServerVariable"}, "type": "object"}, {"type": "null"}], "default": null, "title": "Variables"}}, "required": ["url"], "title": "Server", "type": "object"}, "ServerVariable": {"additionalProperties": true, "properties": {"enum": {"anyOf": [{"items": {"type": "string"}, "type": "array"}, {"type": "null"}], "default": null, "title": "Enum"}, "default": {"title": "Default", "type": "string"}, "description": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": null, "title": "Description"}}, "required": ["default"], "title": "ServerVariable", "type": "object"}}, "securitySchemes": {"bearerTokenAuth": {"type": "http", "scheme": "bearer"}}}}
+{"openapi": "3.1.0", "info": {"title": "", "description": "", "version": "1.0.0"}, "paths": {"/some_endpoint/{param}": {"get": {"summary": "", "description": "", "parameters": [{"required": true, "schema": {"type": "integer"}, "name": "param", "in": "path"}], "requestBody": {"description": "", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Server"}}}, "required": true}, "responses": {"200": {"description": "", "content": {"application/json": {"schema": {"type": "integer"}}}}}, "security": [{"bearerTokenAuth": []}]}}}, "components": {"schemas": {"Server": {"additionalProperties": true, "properties": {"url": {"anyOf": [{"format": "uri", "minLength": 1, "type": "string"}, {"type": "string"}], "title": "Url"}, "description": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": null, "title": "Description"}, "variables": {"anyOf": [{"additionalProperties": {"$ref": "#/components/schemas/ServerVariable"}, "type": "object"}, {"type": "null"}], "default": null, "title": "Variables"}}, "required": ["url"], "title": "Server", "type": "object"}, "ServerVariable": {"additionalProperties": true, "properties": {"enum": {"anyOf": [{"items": {"type": "string"}, "type": "array"}, {"type": "null"}], "default": null, "title": "Enum"}, "default": {"type": "string", "title": "Default"}, "description": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": null, "title": "Description"}}, "required": ["default"], "title": "ServerVariable", "type": "object"}}, "securitySchemes": {"bearerTokenAuth": {"type": "http", "scheme": "bearer"}}}}

tests/unit/test_router.py

@@ -1,4 +1,7 @@
+from typing import Union
+
 import pytest
+from pydantic import BaseModel
 
 from flask_ninja import Query
 from flask_ninja.operation import Callback, Operation
@@ -43,6 +46,51 @@ def sample_method():
     assert router.operations[0].params == [param]
 
 
+def test_add_route_union():
+    router = Router()
+    callback = Callback(
+        name="some_name",
+        url="some_url",
+        method="some_callback_method",
+        response_codes={},
+    )
+
+    param = create_model_field(name="some_param", type_=int, field_info=Query())
+
+    class Response200(BaseModel):
+        status: str
+
+    class Response400(BaseModel):
+        error: str
+
+    @router.add_route(
+        "GET",
+        "/foo",
+        responses={200: Response200, 400: Response400},
+        auth="some_auth",
+        summary="some_summary",
+        description="some_description",
+        params=[param],
+        callbacks=[callback],
+    )
+    def sample_method() -> Union[Response200, Response400]:
+        return Response200(status="foo")
+
+    assert len(router.operations) == 1
+    assert router.operations[0].path == "/foo"
+    assert router.operations[0].method == "GET"
+    assert str(router.operations[0].responses) == str(
+        {
+            200: create_model_field(name="Response 200", type_=Response200),
+            400: create_model_field(name="Response 400", type_=Response400),
+        }
+    )
+    assert router.operations[0].callbacks == [callback]
+    assert router.operations[0].summary == "some_summary"
+    assert router.operations[0].description == "some_description"
+    assert router.operations[0].params == [param]
+
+
 def test_add_route_no_params():
     router = Router()
 

tests/unit/test_router_py3_10.py

@@ -0,0 +1,53 @@
+# type: ignore
+# ^ type check with python 3.9 fails because of "A | B" union syntax
+from pydantic import BaseModel
+
+from flask_ninja import Query
+from flask_ninja.operation import Callback
+from flask_ninja.router import Router
+from flask_ninja.utils import create_model_field
+
+
+def test_add_route_union():
+    router = Router()
+    callback = Callback(
+        name="some_name",
+        url="some_url",
+        method="some_callback_method",
+        response_codes={},
+    )
+
+    param = create_model_field(name="some_param", type_=int, field_info=Query())
+
+    class Response200(BaseModel):
+        status: str
+
+    class Response400(BaseModel):
+        error: str
+
+    @router.add_route(
+        "GET",
+        "/foo",
+        responses={200: Response200, 400: Response400},
+        auth="some_auth",
+        summary="some_summary",
+        description="some_description",
+        params=[param],
+        callbacks=[callback],
+    )
+    def sample_method() -> Response200 | Response400:
+        return Response200(status="foo")
+
+    assert len(router.operations) == 1
+    assert router.operations[0].path == "/foo"
+    assert router.operations[0].method == "GET"
+    assert str(router.operations[0].responses) == str(
+        {
+            200: create_model_field(name="Response 200", type_=Response200),
+            400: create_model_field(name="Response 400", type_=Response400),
+        }
+    )
+    assert router.operations[0].callbacks == [callback]
+    assert router.operations[0].summary == "some_summary"
+    assert router.operations[0].description == "some_description"
+    assert router.operations[0].params == [param]

tox.ini

@@ -1,5 +1,5 @@
 [tox]
-envlist = py3.9, py3.10, py3.11
+envlist = py39, py310, py311
 isolated_build = True
 
 [testenv]
@@ -9,3 +9,10 @@ deps = pytest
 commands =
     # NOTE: you can run any command line tool here - not just tests
     pytest
+
+[testenv:py39]
+deps = pytest
+       pytest-snapshot
+commands =
+    # NOTE: test modules with _py3_10 suffix contain py3.9-incompatible syntax
+    pytest --ignore-glob='*_py3_10.py'