py: added fhirpy support search/fetch API

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

@@ -1,4 +1,4 @@
-from typing import Any, ClassVar, Type, Union, Optional, Iterator, Tuple, Dict
+from typing import Any, Union, Optional, Iterator, Tuple, Dict
 from pydantic import BaseModel, Field
 from typing import Protocol
 
@@ -8,23 +8,21 @@ class ResourceProtocol(Protocol):
     id: Union[str, None]
 
 
-class ResourceTypeDescriptor:
-    def __get__(self, instance: Optional[BaseModel], owner: Type[BaseModel]) -> str:
-        field = owner.model_fields.get("resource_type")
-        if field is None:
-            raise ValueError("resource_type field not found")
-        if field.default is None:
-            raise ValueError("resource_type field default value is not set")
-        return str(field.default)
-
-
 class FhirpyBaseModel(BaseModel):
     """
-    This class satisfies ResourceProtocol
+    This class satisfies ResourceProtocol.
+    Uses __pydantic_init_subclass__ to set resourceType as a class-level attribute
+    after Pydantic finishes model construction, so that fhirpy can detect it
+    via cls.resourceType for search/fetch operations.
     """
     id: Optional[str] = Field(None, alias="id")
 
-    resourceType: ClassVar[ResourceTypeDescriptor] = ResourceTypeDescriptor()
+    @classmethod
+    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
+        super().__pydantic_init_subclass__(**kwargs)
+        field = cls.model_fields.get("resource_type") or cls.model_fields.get("resourceType")
+        if field is not None and field.default is not None:
+            type.__setattr__(cls, "resourceType", str(field.default))
 
     def __iter__(self) -> Iterator[Tuple[str, Any]]:  # type: ignore[override]
         data = self.model_dump(mode='json', by_alias=True, exclude_none=True)

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

@@ -10,10 +10,20 @@ class ResourceProtocol(Protocol):
 
 class FhirpyBaseModel(BaseModel):
     """
-    This class satisfies ResourceProtocol
+    This class satisfies ResourceProtocol.
+    Uses __pydantic_init_subclass__ to set resourceType as a class-level attribute
+    after Pydantic finishes model construction, so that fhirpy can detect it
+    via cls.resourceType for search/fetch operations.
     """
     id: Optional[str] = Field(None, alias="id")
 
+    @classmethod
+    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
+        super().__pydantic_init_subclass__(**kwargs)
+        field = cls.model_fields.get("resource_type") or cls.model_fields.get("resourceType")
+        if field is not None and field.default is not None:
+            type.__setattr__(cls, "resourceType", str(field.default))
+
     def __iter__(self) -> Iterator[Tuple[str, Any]]:  # type: ignore[override]
         data = self.model_dump(mode='json', by_alias=True, exclude_none=True)
         return iter(data.items())

assets/api/writer-generator/python/requirements.txt

@@ -1,5 +1,7 @@
 requests>=2.32.0,<3.0.0
 pytest>=8.3.0,<9.0.0
+pytest-asyncio>=0.24.0,<1.0.0
 pydantic>=2.11.0,<3.0.0
 mypy>=1.9.0,<2.0.0
-types-requests>=2.32.0,<3.0.0
+types-requests>=2.32.0,<3.0.0
+fhirpy>=2.0.0,<3.0.0

examples/python-fhirpy/client.py

@@ -15,40 +15,59 @@
 
 
 async def main() -> None:
-    """
-    Demonstrates usage of fhirpy AsyncFHIRClient to create and fetch FHIR resources.
-    Both Client and Resource APIs are showcased.
-    """
 
     client = AsyncFHIRClient(
         FHIR_SERVER_URL,
         authorization=f"Basic {TOKEN}",
         dump_resource=lambda x: x.model_dump(exclude_none=True),
     )
 
+    # Create a Patient
     patient = Patient(
         name=[HumanName(given=["Bob"], family="Cool2")],
         gender="female",
         birthDate="1980-01-01",
     )
-
-    # Create the Patient using fhirpy's client API
     created_patient = await client.create(patient)
-
     print(f"Created patient: {created_patient.id}")
     print(json.dumps(created_patient.model_dump(exclude_none=True), indent=2))
 
+    # Create an Organization
     organization = Organization(
         name="Beda Software",
         active=True
     )
     created_organization = await client.create(organization)
-
     print(f"Created organization: {created_organization.id}")
 
+    # Search for all patients
     patients = await client.resources(Patient).fetch()
+    print(f"\nFound {len(patients)} patients:")
     for pat in patients:
-        print(f"Found: {pat.name[0].family}")
+        print(f"  - {pat.name[0].family}, {pat.name[0].given[0]}")
+
+    # Search with filters
+    female_patients = await client.resources(Patient).search(gender="female").fetch()
+    print(f"\nFound {len(female_patients)} female patients")
+
+    # Search and limit results
+    first_patient = await client.resources(Patient).first()
+    if first_patient:
+        print(f"\nFirst patient: {first_patient.name[0].family}")
+
+    # Fetch a single patient by ID
+    fetched_patient = await client.reference("Patient", created_patient.id).to_resource()
+    print(f"\nFetched patient by ID: {fetched_patient.name[0].family}")
+
+    # Update a patient
+    created_patient.name = [HumanName(given=["Bob"], family="Updated")]
+    updated_patient = await client.update(created_patient)
+    print(f"\nUpdated patient family name to: {updated_patient.name[0].family}")
+
+    # Cleanup
+    await client.delete(f"Patient/{created_patient.id}")
+    await client.delete(f"Organization/{created_organization.id}")
+    print("\nCleaned up created resources")
 
 
 if __name__ == "__main__":

examples/python-fhirpy/test_sdk.py

@@ -0,0 +1,185 @@
+import asyncio
+import base64
+from typing import AsyncIterator
+
+import pytest
+import pytest_asyncio
+from fhirpy import AsyncFHIRClient
+
+from fhir_types.hl7_fhir_r4_core import HumanName
+from fhir_types.hl7_fhir_r4_core.bundle import Bundle
+from fhir_types.hl7_fhir_r4_core.observation import Observation
+from fhir_types.hl7_fhir_r4_core.patient import Patient
+from fhir_types.hl7_fhir_r4_core.organization import Organization
+from pydantic import ValidationError
+
+FHIR_SERVER_URL = "http://localhost:8080/fhir"
+USERNAME = "root"
+PASSWORD = (
+    "<SECRET>"
+)
+TOKEN = base64.b64encode(f"{USERNAME}:{PASSWORD}".encode()).decode()
+
+
+@pytest.fixture(scope="module")
+def client() -> AsyncFHIRClient:
+    return AsyncFHIRClient(
+        FHIR_SERVER_URL,
+        authorization=f"Basic {TOKEN}",
+        dump_resource=lambda x: x.model_dump(exclude_none=True),
+    )
+
+
+@pytest_asyncio.fixture
+async def created_patient(client: AsyncFHIRClient) -> AsyncIterator[Patient]:
+    patient = Patient(
+        name=[HumanName(given=["Test"], family="FhirpyPatient")],
+        gender="female",
+        birthDate="1980-01-01",
+    )
+    created = await client.create(patient)
+    yield created
+    try:
+        if created.id is not None:
+            await client.delete(f"Patient/{created.id}")
+    except Exception:
+        pass
+
+
+@pytest.mark.asyncio
+async def test_create_patient(client: AsyncFHIRClient) -> None:
+    patient = Patient(
+        name=[HumanName(given=["Create"], family="Test")],
+        gender="female",
+        birthDate="1980-01-01",
+    )
+
+    created = await client.create(patient)
+    assert created.id is not None
+    assert created.name is not None
+    assert created.name[0].family == "Test"
+    assert created.gender == "female"
+
+    await client.delete(f"Patient/{created.id}")
+
+
+@pytest.mark.asyncio
+async def test_search_patients(client: AsyncFHIRClient, created_patient: Patient) -> None:
+    """Test client.resources(Patient).fetch() — requires resourceType class-level access"""
+    patients = await client.resources(Patient).fetch()
+    assert len(patients) > 0
+
+    found = None
+    for p in patients:
+        if p.id == created_patient.id:
+            found = p
+            break
+    assert found is not None, f"Patient {created_patient.id} not found in search results"
+
+
+@pytest.mark.asyncio
+async def test_search_with_filters(client: AsyncFHIRClient, created_patient: Patient) -> None:
+    """Test client.resources(Patient).search(family='FhirpyPatient').fetch()"""
+    patients = await client.resources(Patient).search(family="FhirpyPatient").fetch()
+    assert len(patients) > 0
+
+    ids = [p.id for p in patients]
+    assert created_patient.id in ids
+
+
+@pytest.mark.asyncio
+async def test_search_returns_typed_resources(client: AsyncFHIRClient, created_patient: Patient) -> None:
+    """Verify that fetched resources are deserialized into our generated Patient class"""
+    patients = await client.resources(Patient).fetch()
+    for p in patients:
+        assert isinstance(p, Patient)
+        assert p.resourceType == "Patient"
+
+
+@pytest.mark.asyncio
+async def test_update_patient(client: AsyncFHIRClient, created_patient: Patient) -> None:
+    assert created_patient.id is not None
+
+    created_patient.name = [HumanName(given=["Updated"], family="FhirpyPatient")]
+    created_patient.gender = "male"
+    updated = await client.update(created_patient)
+
+    assert updated.id == created_patient.id
+    assert updated.gender == "male"
+    assert updated.name is not None
+    assert updated.name[0].given == ["Updated"]
+
+
+def test_wrong_resource_type() -> None:
+    json = """
+    {
+      "resourceType" : "Bundle",
+      "id" : "bundle-example",
+      "type" : "searchset",
+      "total" : 3,
+      "link" : [{
+        "relation" : "self",
+        "url" : "https://example.com/base/MedicationRequest?patient=347"
+      }],
+      "entry" : [{
+        "fullUrl" : "https://example.com/base/Patient/3123",
+        "resource" : {
+          "resourceType" : "Weird_Patient",
+          "id" : "3123"
+        },
+        "search" : {
+          "mode" : "match",
+          "score" : 1
+        }
+      }]
+    }
+    """
+    with pytest.raises(ValidationError):
+        Bundle.from_json(json)
+
+
+def test_wrong_fields() -> None:
+    json = """
+    {
+      "resourceType" : "Bundle",
+      "id" : "bundle-example",
+      "type" : "searchset",
+      "total" : 3,
+      "link" : [{
+        "relation" : "self",
+        "url" : "https://example.com/base/MedicationRequest?patient=347"
+      }],
+      "entry" : [{
+        "fullUrl" : "https://example.com/base/Patient/3123",
+        "resource" : {
+          "resourceType" : "Patient",
+          "id" : "3123",
+          "very_wrong_field" : "WRONG"
+        },
+        "search" : {
+          "mode" : "match",
+          "score" : 1
+        }
+      }]
+    }
+    """
+    with pytest.raises(ValidationError):
+        Bundle.from_json(json)
+
+
+def test_to_from_json() -> None:
+    p = Patient(
+        name=[HumanName(given=["Test"], family="Patient")],
+        gender="female",
+        birthDate="1980-01-01",
+    )
+    json = p.to_json(indent=2)
+    p2 = Patient.from_json(json)
+    assert p == p2
+
+
+def test_resource_type_class_access() -> None:
+    """Verify that resourceType is accessible at class level (needed for fhirpy search)"""
+    assert Patient.resourceType == "Patient"
+    assert Observation.resourceType == "Observation"
+    assert Bundle.resourceType == "Bundle"

examples/python/fhir_types/requirements.txt

@@ -1,5 +1,7 @@
 requests>=2.32.0,<3.0.0
 pytest>=8.3.0,<9.0.0
+pytest-asyncio>=0.24.0,<1.0.0
 pydantic>=2.11.0,<3.0.0
 mypy>=1.9.0,<2.0.0
-types-requests>=2.32.0,<3.0.0
+types-requests>=2.32.0,<3.0.0
+fhirpy>=2.0.0,<3.0.0

test/api/write-generator/__snapshots__/python.test.ts.snap

@@ -78,7 +78,9 @@ class Patient(DomainResource):
 exports[`Python Writer Generator static files 1`] = `
 "requests>=2.32.0,<3.0.0
 pytest>=8.3.0,<9.0.0
+pytest-asyncio>=0.24.0,<1.0.0
 pydantic>=2.11.0,<3.0.0
 mypy>=1.9.0,<2.0.0
-types-requests>=2.32.0,<3.0.0"
+types-requests>=2.32.0,<3.0.0
+fhirpy>=2.0.0,<3.0.0"
 `;