py: generics

Makefile

@@ -156,6 +156,10 @@ test-python-sdk: typecheck prepare-aidbox-runme generate-python-sdk python-test-
          . venv/bin/activate && \
          python -m pytest test_raw_extension.py -v
 
+	cd $(PYTHON_EXAMPLE) && \
+         . venv/bin/activate && \
+         python -m pytest test_bundle.py -v
+
 test-python-fhirpy-sdk: typecheck prepare-aidbox-runme generate-python-sdk-fhirpy python-fhirpy-test-setup
     # Run mypy in strict mode
 	cd $(PYTHON_FHIRPY_EXAMPLE) && \

assets/api/writer-generator/python/resource_preprocessor.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import re
+import importlib
+import importlib.util
+from typing import Any
+
+
+def _to_snake_case(name: str) -> str:
+    return re.sub(r"(?<!^)(?=[A-Z])", "_", name).lower()
+
+
+def _import_resource_class(package: str, resource_type: str) -> Any:
+    module_name = f"{package}.{_to_snake_case(resource_type)}"
+    if importlib.util.find_spec(module_name) is None:
+        return None
+    module = importlib.import_module(module_name)
+    return getattr(module, resource_type, None)
+
+
+def _preprocess_value(value: Any, package: str) -> Any:
+    if isinstance(value, dict):
+        resource_type = value.get("resourceType")
+        if resource_type and isinstance(resource_type, str):
+            cls = _import_resource_class(package, resource_type)
+            if cls is not None:
+                return cls.model_validate(value)
+        return {k: _preprocess_value(v, package) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_preprocess_value(item, package) for item in value]
+    return value
+
+
+def preprocess_resource_fields(data: dict[str, Any], package: str) -> dict[str, Any]:
+    """Walk a FHIR resource dict and replace nested resource dicts with concrete model instances.
+
+    Intended for use as a model_validator(mode='before') on generic resource containers
+    such as Bundle or DomainResource. Processes field values (not the root dict itself) so
+    the caller's own Pydantic validation still runs normally.
+    """
+    return {k: _preprocess_value(v, package) for k, v in data.items()}

examples/python/fhir_types/__init__.py

@@ -13,7 +13,6 @@
     Bundle, BundleEntry, BundleEntryRequest, BundleEntryResponse, BundleEntrySearch, BundleLink
 )
 from fhir_types.hl7_fhir_r4_core.domain_resource import DomainResource
-from fhir_types.hl7_fhir_r4_core.resource_families import DomainResourceFamily
 from fhir_types.hl7_fhir_r4_core.observation import (\
     Observation, ObservationComponent, ObservationReferenceRange
 )
@@ -22,7 +21,6 @@
     Patient, PatientCommunication, PatientContact, PatientLink
 )
 from fhir_types.hl7_fhir_r4_core.resource import Resource
-from fhir_types.hl7_fhir_r4_core.resource_families import ResourceFamily
 
 Address.model_rebuild()
 Age.model_rebuild()

examples/python/fhir_types/hl7_fhir_r4_core/__init__.py

@@ -13,7 +13,6 @@
     Bundle, BundleEntry, BundleEntryRequest, BundleEntryResponse, BundleEntrySearch, BundleLink
 )
 from fhir_types.hl7_fhir_r4_core.domain_resource import DomainResource
-from fhir_types.hl7_fhir_r4_core.resource_families import DomainResourceFamily
 from fhir_types.hl7_fhir_r4_core.observation import (\
     Observation, ObservationComponent, ObservationReferenceRange
 )
@@ -22,7 +21,6 @@
     Patient, PatientCommunication, PatientContact, PatientLink
 )
 from fhir_types.hl7_fhir_r4_core.resource import Resource
-from fhir_types.hl7_fhir_r4_core.resource_families import ResourceFamily
 
 __all__ = [
     'Address',

examples/python/fhir_types/hl7_fhir_r4_core/bundle.py

@@ -3,24 +3,36 @@
 # Any manual changes made to this file may be overwritten.
 
 from __future__ import annotations
-from pydantic import BaseModel, ConfigDict, Field, PositiveInt
-from typing import Any, List as PyList, Literal
+from pydantic import BaseModel, ConfigDict, Field, PositiveInt, model_validator
+from typing import Any, Generic, List as PyList, Literal
+from typing_extensions import Self, TypeVar
 
 from fhir_types.hl7_fhir_r4_core.base import BackboneElement, Identifier, Signature
 from fhir_types.hl7_fhir_r4_core.resource import Resource
-from fhir_types.hl7_fhir_r4_core.resource_families import ResourceFamily
 from fhir_types.hl7_fhir_r4_core.base import Element
+from fhir_types.hl7_fhir_r4_core.resource_preprocessor import preprocess_resource_fields
 
+T = TypeVar('T', bound=Resource, default=Resource)
+T1 = TypeVar('T1', bound=Resource, default=Resource)
+T2 = TypeVar('T2', bound=Resource, default=Resource)
 
-class BundleEntry(BackboneElement):
+
+class BundleEntry(BackboneElement, Generic[T1, T2]):
     model_config = ConfigDict(validate_by_name=True, serialize_by_alias=True, extra="forbid")
     full_url: str | None = Field(None, alias="fullUrl", serialization_alias="fullUrl")
     link: PyList[BundleLink] | None = Field(None, alias="link", serialization_alias="link")
     request: BundleEntryRequest | None = Field(None, alias="request", serialization_alias="request")
-    resource: ResourceFamily | None = Field(None, alias="resource", serialization_alias="resource")
-    response: BundleEntryResponse | None = Field(None, alias="response", serialization_alias="response")
+    resource: T1 | None = Field(None, alias="resource", serialization_alias="resource")
+    response: BundleEntryResponse[T2] | None = Field(None, alias="response", serialization_alias="response")
     search: BundleEntrySearch | None = Field(None, alias="search", serialization_alias="search")
 
+    @model_validator(mode='before')
+    @classmethod
+    def _preprocess_resources(cls, data: Any) -> Any:
+        if isinstance(data, dict):
+            return preprocess_resource_fields(data, "fhir_types.hl7_fhir_r4_core")
+        return data
+
 class BundleEntryRequest(BackboneElement):
     model_config = ConfigDict(validate_by_name=True, serialize_by_alias=True, extra="forbid")
     if_match: str | None = Field(None, alias="ifMatch", serialization_alias="ifMatch")
@@ -30,14 +42,21 @@ class BundleEntryRequest(BackboneElement):
     method: Literal["GET", "HEAD", "POST", "PUT", "DELETE", "PATCH"] = Field(alias="method", serialization_alias="method")
     url: str = Field(alias="url", serialization_alias="url")
 
-class BundleEntryResponse(BackboneElement):
+class BundleEntryResponse(BackboneElement, Generic[T]):
     model_config = ConfigDict(validate_by_name=True, serialize_by_alias=True, extra="forbid")
     etag: str | None = Field(None, alias="etag", serialization_alias="etag")
     last_modified: str | None = Field(None, alias="lastModified", serialization_alias="lastModified")
     location: str | None = Field(None, alias="location", serialization_alias="location")
-    outcome: ResourceFamily | None = Field(None, alias="outcome", serialization_alias="outcome")
+    outcome: T | None = Field(None, alias="outcome", serialization_alias="outcome")
     status: str = Field(alias="status", serialization_alias="status")
 
+    @model_validator(mode='before')
+    @classmethod
+    def _preprocess_resources(cls, data: Any) -> Any:
+        if isinstance(data, dict):
+            return preprocess_resource_fields(data, "fhir_types.hl7_fhir_r4_core")
+        return data
+
 class BundleEntrySearch(BackboneElement):
     model_config = ConfigDict(validate_by_name=True, serialize_by_alias=True, extra="forbid")
     mode: Literal["match", "include", "outcome"] | None = Field(None, alias="mode", serialization_alias="mode")
@@ -49,7 +68,7 @@ class BundleLink(BackboneElement):
     url: str = Field(alias="url", serialization_alias="url")
 
 
-class Bundle(Resource):
+class Bundle(Resource, Generic[T1, T2]):
     model_config = ConfigDict(validate_by_name=True, serialize_by_alias=True, extra="forbid")
     resource_type: Literal['Bundle'] = Field(
         default='Bundle',
@@ -58,7 +77,7 @@ class Bundle(Resource):
         frozen=True,
         pattern='Bundle'
     )
-    entry: PyList[BundleEntry] | None = Field(None, alias="entry", serialization_alias="entry")
+    entry: PyList[BundleEntry[T1, T2]] | None = Field(None, alias="entry", serialization_alias="entry")
     identifier: Identifier | None = Field(None, alias="identifier", serialization_alias="identifier")
     link: PyList[BundleLink] | None = Field(None, alias="link", serialization_alias="link")
     signature: Signature | None = Field(None, alias="signature", serialization_alias="signature")
@@ -76,6 +95,13 @@ def to_json(self, indent: int | None = None) -> str:
         return self.model_dump_json(exclude_unset=True, exclude_none=True, indent=indent)
 
     @classmethod
-    def from_json(cls, json: str) -> Bundle:
+    def from_json(cls, json: str) -> Self:
         return cls.model_validate_json(json)
 
+    @model_validator(mode='before')
+    @classmethod
+    def _preprocess_resources(cls, data: Any) -> Any:
+        if isinstance(data, dict):
+            return preprocess_resource_fields(data, "fhir_types.hl7_fhir_r4_core")
+        return data
+

examples/python/fhir_types/hl7_fhir_r4_core/domain_resource.py

@@ -3,15 +3,18 @@
 # Any manual changes made to this file may be overwritten.
 
 from __future__ import annotations
-from pydantic import BaseModel, ConfigDict, Field, PositiveInt
-from typing import Any, List as PyList, Literal
+from pydantic import BaseModel, ConfigDict, Field, PositiveInt, model_validator
+from typing import Any, Generic, List as PyList, Literal
+from typing_extensions import Self, TypeVar
 
 from fhir_types.hl7_fhir_r4_core.base import Extension, Narrative
 from fhir_types.hl7_fhir_r4_core.resource import Resource
-from fhir_types.hl7_fhir_r4_core.resource_families import ResourceFamily
+from fhir_types.hl7_fhir_r4_core.resource_preprocessor import preprocess_resource_fields
 
+T = TypeVar('T', bound=Resource, default=Resource)
 
-class DomainResource(Resource):
+
+class DomainResource(Resource, Generic[T]):
     model_config = ConfigDict(validate_by_name=True, serialize_by_alias=True, extra="forbid")
     resource_type: str = Field(
         default='DomainResource',
@@ -20,7 +23,7 @@ class DomainResource(Resource):
         frozen=True,
         pattern='DomainResource'
     )
-    contained: PyList[ResourceFamily] | None = Field(None, alias="contained", serialization_alias="contained")
+    contained: PyList[T] | None = Field(None, alias="contained", serialization_alias="contained")
     extension: PyList[Extension] | None = Field(None, alias="extension", serialization_alias="extension")
     modifier_extension: PyList[Extension] | None = Field(None, alias="modifierExtension", serialization_alias="modifierExtension")
     text: Narrative | None = Field(None, alias="text", serialization_alias="text")
@@ -32,6 +35,13 @@ def to_json(self, indent: int | None = None) -> str:
         return self.model_dump_json(exclude_unset=True, exclude_none=True, indent=indent)
 
     @classmethod
-    def from_json(cls, json: str) -> DomainResource:
+    def from_json(cls, json: str) -> Self:
         return cls.model_validate_json(json)
 
+    @model_validator(mode='before')
+    @classmethod
+    def _preprocess_resources(cls, data: Any) -> Any:
+        if isinstance(data, dict):
+            return preprocess_resource_fields(data, "fhir_types.hl7_fhir_r4_core")
+        return data
+

examples/python/fhir_types/hl7_fhir_r4_core/observation.py

@@ -5,13 +5,13 @@
 from __future__ import annotations
 from pydantic import BaseModel, ConfigDict, Field, PositiveInt
 from typing import Any, List as PyList, Literal
+from typing_extensions import Self
 
 from fhir_types.hl7_fhir_r4_core.base import (\
     Annotation, BackboneElement, CodeableConcept, Identifier, Period, Quantity, Range, Ratio, Reference, SampledData, \
     Timing
 )
 from fhir_types.hl7_fhir_r4_core.domain_resource import DomainResource
-from fhir_types.hl7_fhir_r4_core.resource_families import DomainResourceFamily
 from fhir_types.hl7_fhir_r4_core.base import Element
 
 
@@ -106,6 +106,6 @@ def to_json(self, indent: int | None = None) -> str:
         return self.model_dump_json(exclude_unset=True, exclude_none=True, indent=indent)
 
     @classmethod
-    def from_json(cls, json: str) -> Observation:
+    def from_json(cls, json: str) -> Self:
         return cls.model_validate_json(json)
 

examples/python/fhir_types/hl7_fhir_r4_core/operation_outcome.py

@@ -5,10 +5,10 @@
 from __future__ import annotations
 from pydantic import BaseModel, ConfigDict, Field, PositiveInt
 from typing import Any, List as PyList, Literal
+from typing_extensions import Self
 
 from fhir_types.hl7_fhir_r4_core.base import BackboneElement, CodeableConcept
 from fhir_types.hl7_fhir_r4_core.domain_resource import DomainResource
-from fhir_types.hl7_fhir_r4_core.resource_families import DomainResourceFamily
 
 
 class OperationOutcomeIssue(BackboneElement):
@@ -39,6 +39,6 @@ def to_json(self, indent: int | None = None) -> str:
         return self.model_dump_json(exclude_unset=True, exclude_none=True, indent=indent)
 
     @classmethod
-    def from_json(cls, json: str) -> OperationOutcome:
+    def from_json(cls, json: str) -> Self:
         return cls.model_validate_json(json)
 

examples/python/fhir_types/hl7_fhir_r4_core/patient.py

@@ -5,12 +5,12 @@
 from __future__ import annotations
 from pydantic import BaseModel, ConfigDict, Field, PositiveInt
 from typing import Any, List as PyList, Literal
+from typing_extensions import Self
 
 from fhir_types.hl7_fhir_r4_core.base import (\
     Address, Attachment, BackboneElement, CodeableConcept, ContactPoint, HumanName, Identifier, Period, Reference
 )
 from fhir_types.hl7_fhir_r4_core.domain_resource import DomainResource
-from fhir_types.hl7_fhir_r4_core.resource_families import DomainResourceFamily
 from fhir_types.hl7_fhir_r4_core.base import Element
 
 
@@ -77,6 +77,6 @@ def to_json(self, indent: int | None = None) -> str:
         return self.model_dump_json(exclude_unset=True, exclude_none=True, indent=indent)
 
     @classmethod
-    def from_json(cls, json: str) -> Patient:
+    def from_json(cls, json: str) -> Self:
         return cls.model_validate_json(json)