Merge pull request #22 from ffkirill/support_OptionalUnionT

Fix bug optional parameter is reported as required.

aiohttp_pydantic/oas/view.py

@@ -16,21 +16,6 @@ from ..view import PydanticView, is_pydantic_view
 from .typing import is_status_code_type
 
 
-def _handle_optional(type_):
-    """
-    Returns the type wrapped in Optional or None.
-    >>>  from typing import Optional
-    >>>  _handle_optional(int)
-    >>>  _handle_optional(Optional[str])
-    <class 'str'>
-    """
-    if typing.get_origin(type_) is typing.Union:
-        args = typing.get_args(type_)
-        if len(args) == 2 and type(None) in args:
-            return next(iter(set(args) - {type(None)}))
-    return None
-
-
 class _OASResponseBuilder:
     """
     Parse the type annotated as returned by a function and
@@ -128,18 +113,18 @@ def _add_http_method_to_oas(
             i = next(indexes)
             oas_operation.parameters[i].in_ = args_location
             oas_operation.parameters[i].name = name
-            optional_type = _handle_optional(type_)
 
             attrs = {"__annotations__": {"__root__": type_}}
             if name in defaults:
                 attrs["__root__"] = defaults[name]
+                oas_operation.parameters[i].required = False
+            else:
+                oas_operation.parameters[i].required = True
 
             oas_operation.parameters[i].schema = type(name, (BaseModel,), attrs).schema(
                 ref_template="#/components/schemas/{model}"
             )
 
-            oas_operation.parameters[i].required = optional_type is None
-
     return_type = handler.__annotations__.get("return")
     if return_type is not None:
         _OASResponseBuilder(oas, oas_operation, status_code_descriptions).build(

tests/test_oas/test_view.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import List, Optional, Union
+from typing import List, Optional, Union, Literal
 from uuid import UUID
 
 import pytest
@@ -46,7 +46,7 @@ class PetCollectionView(PydanticView):
 
 
 class PetItemView(PydanticView):
-    async def get(self, id: int, /) -> Union[r200[Pet], r404]:
+    async def get(self, id: int, /, size: Union[int, Literal['x', 'l', 's']], day: Union[int, Literal["now"]] = "now") -> Union[r200[Pet], r404]:
         return web.json_response()
 
     async def put(self, id: int, /, pet: Pet):
@@ -245,7 +245,21 @@ async def test_pets_id_route_should_have_get_method(generated_oas):
                 "name": "id",
                 "required": True,
                 "schema": {"title": "id", "type": "integer"},
-            }
+            },
+            {'in': 'query',
+             'name': 'size',
+             'required': True,
+             'schema': {'anyOf': [{'type': 'integer'},
+                                  {'enum': ['x', 'l', 's'],
+                                   'type': 'string'}],
+                        'title': 'size'}},
+            {'in': 'query',
+             'name': 'day',
+             'required': False,
+             'schema': {'anyOf': [{'type': 'integer'},
+                                  {'enum': ['now'], 'type': 'string'}],
+                        'default': 'now',
+                        'title': 'day'}}
         ],
         "responses": {
             "200": {