Support union deserialization for OpenAPI models

examples/EmergencyManagement/Server/models_emergency.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
-from typing import Optional
+
+from typing import Union, Optional
 
 from reticulum_openapi.model import BaseModel
 from sqlalchemy.orm import declarative_base
@@ -86,3 +87,29 @@ class Event(BaseModel):
     detail: Optional[Detail] = None
     point: Optional[Point] = None
     __orm_model__ = EventORM
+
+
+# --- Additional example models demonstrating allOf/oneOf/anyOf ---
+
+@dataclass
+class BaseVehicle(BaseModel):
+    manufacturer: str
+
+
+@dataclass
+class Car(BaseVehicle):
+    doors: int
+
+
+@dataclass
+class Bike(BaseModel):
+    handlebar: str
+
+
+Vehicle = Union[Car, Bike]
+
+
+@dataclass
+class TransportRecord(BaseModel):
+    owner: str
+    vehicle: Vehicle

reticulum_openapi/model.py

@@ -2,7 +2,7 @@
 from dataclasses import dataclass, asdict, is_dataclass, fields
 import json
 import zlib
-from typing import Type, TypeVar
+from typing import Type, TypeVar, get_origin, get_args, Union
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
 from sqlalchemy import select
 
@@ -45,8 +45,28 @@ def dataclass_from_json(cls: Type[T], data: bytes) -> T:
         json_bytes = data
     json_str = json_bytes.decode('utf-8')
     obj_dict = json.loads(json_str)
-    # Instantiate dataclass by unpacking dict (assumes keys match field names)
-    return cls(**obj_dict)  # type: ignore
+
+    def _construct(tp, value):
+        origin = get_origin(tp)
+        if origin is Union:
+            for sub in get_args(tp):
+                try:
+                    return _construct(sub, value)
+                except Exception:
+                    continue
+            raise ValueError(f"No matching type for Union {tp}")
+        if is_dataclass(tp):
+            kwargs = {}
+            for f in fields(tp):
+                if isinstance(value, dict) and f.name in value:
+                    kwargs[f.name] = _construct(f.type, value[f.name])
+            return tp(**kwargs)  # type: ignore
+        if origin is list and isinstance(value, list):
+            item_type = get_args(tp)[0]
+            return [_construct(item_type, v) for v in value]
+        return value
+
+    return _construct(cls, obj_dict)
 
 
 @dataclass

tests/test_model.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from typing import Union
 from reticulum_openapi.model import dataclass_to_json, dataclass_from_json
 
 
@@ -13,3 +14,41 @@ def test_serialization_roundtrip():
     data = dataclass_to_json(item)
     obj = dataclass_from_json(Item, data)
     assert obj == item
+
+
+@dataclass
+class BaseVehicle:
+    manufacturer: str
+
+
+@dataclass
+class Car(BaseVehicle):
+    doors: int
+
+
+@dataclass
+class Bike:
+    handlebar: str
+
+
+Vehicle = Union[Car, Bike]
+
+
+@dataclass
+class TransportRecord:
+    owner: str
+    vehicle: Vehicle
+
+
+def test_union_deserialization_root():
+    data = dataclass_to_json(Car(manufacturer="Acme", doors=2))
+    obj = dataclass_from_json(Vehicle, data)
+    assert isinstance(obj, Car)
+    assert obj.doors == 2
+
+
+def test_union_deserialization_nested():
+    record = TransportRecord(owner="bob", vehicle=Bike(handlebar="drop"))
+    data = dataclass_to_json(record)
+    obj = dataclass_from_json(TransportRecord, data)
+    assert isinstance(obj.vehicle, Bike)