Merge pull request #1690 from vitalik/uniont-type

Optional Union fix

Author: vitalik

ninja/signature/details.py

@@ -208,6 +208,7 @@ def _args_flatten_map(self, args: List[FuncParam]) -> Dict[str, Tuple[str, ...]]
         return flatten_map
 
     def _model_flatten_map(self, model: TModel, prefix: str) -> Generator:
+        model = _unwrap_union_model(model)
         field: FieldInfo
         for attr, field in model.model_fields.items():
             field_name = field.alias or attr
@@ -306,6 +307,15 @@ def _get_param_type(self, name: str, arg: inspect.Parameter) -> FuncParam:
         )
 
 
+def _unwrap_union_model(annotation: Any) -> Any:
+    """If annotation is a Union containing a pydantic model, return that model class."""
+    if get_origin(annotation) in UNION_TYPES:
+        for arg in get_args(annotation):
+            if arg is not type(None) and is_pydantic_model(arg):
+                return arg
+    return annotation
+
+
 def is_pydantic_model(cls: Any) -> bool:
     try:
         origin = get_origin(cls)
@@ -360,6 +370,7 @@ def detect_collection_fields(
             for attr in path[1:]:
                 if hasattr(annotation_or_field, "annotation"):
                     annotation_or_field = annotation_or_field.annotation
+                annotation_or_field = _unwrap_union_model(annotation_or_field)
                 annotation_or_field = next(
                     (
                         a

tests/test_query_schema.py

@@ -1,11 +1,16 @@
+import sys
 from datetime import datetime
 from enum import IntEnum
+from typing import Optional
 
+import pytest
 from pydantic import BaseModel, Field
 
 from ninja import NinjaAPI, Query, Schema
 from ninja.testing.client import TestClient
 
+PY_310 = sys.version_info >= (3, 10)
+
 
 class Range(IntEnum):
     TWENTY = 20
@@ -163,3 +168,87 @@ def test_schema_all_of_no_ref():
         "default": 1,
         "allOf": [{"title": "Best Type Ever!"}, {"no-ref-here": "xyzzy"}],
     }
+
+
+@pytest.mark.skipif(not PY_310, reason="requires Python 3.10+ pipe syntax")
+def test_unwrap_union_model_no_model():
+    """_unwrap_union_model returns input as-is when union has no pydantic model."""
+    from ninja.signature.details import _unwrap_union_model
+
+    annotation = int | None
+    assert _unwrap_union_model(annotation) is annotation
+    assert _unwrap_union_model(str) is str
+
+
+def test_optional_query_schema():
+    """Optional[Model] in Query should not crash (issue #1634)."""
+
+    class MyFilter(Schema):
+        name: str = "default"
+
+    temp_api = NinjaAPI()
+
+    @temp_api.get("/opt")
+    def view(request, f: Optional[MyFilter] = Query(None)):
+        if f:
+            return f.model_dump()
+        return {}
+
+    client = TestClient(temp_api)
+
+    resp = client.get("/opt?name=hello")
+    assert resp.status_code == 200
+    assert resp.json() == {"name": "hello"}
+
+    resp = client.get("/opt")
+    assert resp.status_code == 200
+
+
+@pytest.mark.skipif(not PY_310, reason="requires Python 3.10+ pipe syntax")
+def test_union_pipe_syntax_query_schema():
+    """Model | None in Query should not crash (issue #1634, Python 3.10+ pipe syntax)."""
+
+    class MyFilter(Schema):
+        name: str = "default"
+
+    temp_api = NinjaAPI()
+
+    @temp_api.get("/pipe")
+    def view(request, f: MyFilter | None = Query(None)):
+        if f:
+            return f.model_dump()
+        return {}
+
+    client = TestClient(temp_api)
+
+    resp = client.get("/pipe?name=world")
+    assert resp.status_code == 200
+    assert resp.json() == {"name": "world"}
+
+    resp = client.get("/pipe")
+    assert resp.status_code == 200
+
+
+@pytest.mark.skipif(not PY_310, reason="requires Python 3.10+ pipe syntax")
+def test_nested_optional_query_schema():
+    """Nested optional model fields should also work."""
+
+    class Inner(Schema):
+        value: Optional[int] = 0
+        items: list[str] = []
+
+    class Outer(Schema):
+        inner: Inner | None = None
+        label: str = "x"
+
+    temp_api = NinjaAPI()
+
+    @temp_api.get("/nested")
+    def view(request, f: Outer = Query(...)):
+        return f.model_dump()
+
+    client = TestClient(temp_api)
+
+    resp = client.get("/nested?label=test")
+    assert resp.status_code == 200
+    assert resp.json()["label"] == "test"