Merge pull request #104 from icanbwell/separate-bundle-resources-when-streaming

fix bug when there is empty bundle and we ask to expand it

Author: imranq2

README.md

@@ -1,3 +1,5 @@
+from typing_extensions import Optionalfrom helix_fhir_client_sdk.responses.fhir_get_response import FhirGetResponsefrom build.lib.helix_fhir_client_sdk.responses.fhir_get_response import FhirGetResponsefrom helix_fhir_client_sdk.responses.fhir_get_response import FhirGetResponse
+
 # helix.fhir.client.sdk
 
 <p align="left">
@@ -40,3 +42,40 @@ https://github.com/icanbwell/fhir-server-performance
 * 1.x supports python 3.7+ 
 * 2.x supports python 3.10+
 * 3.x supports python 3.12+
+
+# Asynchronous Support
+When communicating with FHIR servers, a lot of time is spent waiting for the server to respond. 
+This is a good use case for using asynchronous programming. 
+This SDK supports asynchronous programming using the `async` and `await` keywords.
+
+The return types are Python AsyncGenerators.  Python makes it very easy to work with AsyncGenerators.
+
+For example, if the SDK provides a function like this:
+```python
+
+async def get_resources(self) -> AsyncGenerator[FhirGetResponse, None]:
+    ...
+```
+
+You can iterate over the results as they become available:
+```python
+response: Optional[FhirGetResponse]
+async for response in client.get_resources():
+    print(response.resource)
+```
+
+Or you can get a list of responses (which will return AFTER all the responses are received:
+```python
+
+responses: List[FhirGetResponse] = [response async for response in client.get_resources()]
+```
+
+Or you can aggregate the responses into one response (which will return AFTER all the responses are received:
+```python
+
+response: Optional[FhirGetResponse] = await FhirGetResponse.from_async_generator(client.get_resources())
+```
+
+# Data Streaming
+For FHIR servers that support data streaming (e.g., b.well FHIR server), you can just set the `use_data_streaming` parameter to stream the data as it i received.
+The data will be streamed in AsyncGenerators as described above.

helix_fhir_client_sdk/fhir_merge_mixin.py

@@ -297,7 +297,7 @@ def merge(
         id_: Optional[str] = None,
         json_data_list: List[str],
         batch_size: Optional[int] = None,
-    ) -> FhirMergeResponse:
+    ) -> Optional[FhirMergeResponse]:
         """
         Calls $merge function on FHIR server
 
@@ -308,7 +308,7 @@ def merge(
         :return: response
         """
 
-        result: FhirMergeResponse = AsyncRunner.run(
+        result: Optional[FhirMergeResponse] = AsyncRunner.run(
             FhirMergeResponse.from_async_generator(
                 self.merge_async(
                     id_=id_, json_data_list=json_data_list, batch_size=batch_size

helix_fhir_client_sdk/graph/fhir_graph_mixin.py

@@ -69,7 +69,7 @@ def graph(
         *,
         graph_definition: GraphDefinition,
         contained: bool,
-    ) -> FhirGetResponse:
+    ) -> Optional[FhirGetResponse]:
 
         return AsyncRunner.run(
             FhirGetResponse.from_async_generator(

helix_fhir_client_sdk/graph/simulated_graph_processor_mixin.py

@@ -494,7 +494,7 @@ async def _get_resources_by_parameters_async(
                         )
 
             if cached_response:
-                result.append([cached_response])
+                result.append(cached_response)
         elif cached_response:
             result = cached_response
         assert result
@@ -540,7 +540,7 @@ async def simulate_graph_async(
             assert self._additional_parameters is not None
             self._additional_parameters.append("contained=true")
 
-        result: FhirGetResponse = await FhirGetResponse.from_async_generator(
+        result: Optional[FhirGetResponse] = await FhirGetResponse.from_async_generator(
             self.process_simulate_graph_async(
                 id_=id_,
                 graph_json=graph_json,
@@ -559,6 +559,7 @@ async def simulate_graph_async(
                 auth_scopes=self._auth_scopes,
             )
         )
+        assert result, "No result returned from simulate_graph_async"
         return result
 
     # noinspection PyPep8Naming

helix_fhir_client_sdk/responses/fhir_delete_response.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, AsyncGenerator
 
 
 class FhirDeleteResponse:
@@ -33,3 +33,32 @@ def __init__(
         """ Number of resources deleted """
         self.count: Optional[int] = count
         self.resource_type: Optional[str] = resource_type
+
+    def append(self, other: Optional["FhirDeleteResponse"]) -> None:
+        """
+        Appends another FhirDeleteResponse to this one
+
+        :param other: FhirDeleteResponse to append
+        """
+        if other:
+            self.responses += other.responses
+            self.error = (self.error or "") + (other.error or "")
+            self.count = (self.count or 0) + (other.count or 0)
+
+    @classmethod
+    async def from_async_generator(
+        cls, generator: AsyncGenerator["FhirDeleteResponse", None]
+    ) -> Optional["FhirDeleteResponse"]:
+        """
+        Reads a generator of FhirDeleteResponse and returns a single FhirDeleteResponse by appending all the FhirDeleteResponse
+
+        :param generator: generator of FhirDeleteResponse items
+        :return: FhirDeleteResponse
+        """
+        result: FhirDeleteResponse | None = None
+        async for value in generator:
+            if not result:
+                result = value
+            else:
+                result.append(value)
+        return result

helix_fhir_client_sdk/responses/fhir_get_response.py

@@ -87,7 +87,35 @@ def __init__(
         self.chunk_number: Optional[int] = chunk_number
         self.cache_hits: Optional[int] = cache_hits
 
-    def append(self, others: List["FhirGetResponse"]) -> "FhirGetResponse":
+    def append(self, other_response: "FhirGetResponse") -> "FhirGetResponse":
+        """
+        Append the responses from other to self
+
+        :param other_response: FhirGetResponse object to append to current one
+        :return: self
+        """
+        bundle_entries: List[BundleEntry] = self.get_bundle_entries()
+        if other_response.responses:
+            other_bundle_entries: List[BundleEntry] = (
+                other_response.get_bundle_entries()
+            )
+            bundle_entries.extend(other_bundle_entries)
+        bundle = {
+            "resourceType": "Bundle",
+            "entry": bundle_entries,
+        }
+        self.responses = json.dumps(bundle, cls=FhirJSONEncoder)
+        if other_response.chunk_number and (other_response.chunk_number or 0) > (
+            self.chunk_number or 0
+        ):
+            self.chunk_number = other_response.chunk_number
+        if other_response.next_url:
+            self.next_url = other_response.next_url
+            self.access_token = other_response.access_token
+        self.cache_hits = (self.cache_hits or 0) + (other_response.cache_hits or 0)
+        return self
+
+    def extend(self, others: List["FhirGetResponse"]) -> "FhirGetResponse":
         """
         Append the responses from other to self
 
@@ -373,10 +401,10 @@ def get_resource_type_and_ids(self) -> List[str]:
                 f"Could not get resourceType and id from resources: {json.dumps(resources, cls=FhirJSONEncoder)}"
             ) from e
 
-    @staticmethod
+    @classmethod
     async def from_async_generator(
-        generator: AsyncGenerator["FhirGetResponse", None]
-    ) -> "FhirGetResponse":
+        cls, generator: AsyncGenerator["FhirGetResponse", None]
+    ) -> Optional["FhirGetResponse"]:
         """
         Reads a generator of FhirGetResponse and returns a single FhirGetResponse by appending all the FhirGetResponse
 
@@ -388,7 +416,7 @@ async def from_async_generator(
             if not result:
                 result = value
             else:
-                result.append([value])
+                result.append(value)
 
         assert result
         return result

helix_fhir_client_sdk/responses/fhir_merge_response.py

@@ -30,22 +30,21 @@ def __init__(
         self.data: str = json_data
         self.successful: bool = status != 200
 
-    def append(self, other: Optional[List["FhirMergeResponse"]]) -> None:
+    def append(self, other: Optional["FhirMergeResponse"]) -> None:
         """
         Appends another FhirMergeResponse to this one
 
         :param other: FhirMergeResponse to append
         """
         if other:
-            for r in other:
-                self.responses.extend(r.responses)
-                self.error = (self.error or "") + (r.error or "")
-                self.successful = self.successful and r.successful
+            self.responses.extend(other.responses)
+            self.error = (self.error or "") + (other.error or "")
+            self.successful = self.successful and other.successful
 
-    @staticmethod
+    @classmethod
     async def from_async_generator(
-        generator: AsyncGenerator["FhirMergeResponse", None]
-    ) -> "FhirMergeResponse":
+        cls, generator: AsyncGenerator["FhirMergeResponse", None]
+    ) -> Optional["FhirMergeResponse"]:
         """
         Reads a generator of FhirGetResponse and returns a single FhirGetResponse by appending all the FhirGetResponse
 
@@ -57,7 +56,7 @@ async def from_async_generator(
             if not result:
                 result = value
             else:
-                result.append([value])
+                result.append(value)
 
         assert result
         return result

helix_fhir_client_sdk/responses/fhir_response_processor.py

@@ -387,7 +387,6 @@ async def _handle_response_200_non_streaming(
                         extra_context_to_return=extra_context_to_return,
                         resource_or_bundle=response_json,
                         separate_bundle_resources=separate_bundle_resources,
-                        text=text,
                         total_count=total_count,
                         url=url,
                     )
@@ -434,7 +433,6 @@ async def expand_or_separate_bundle_async(
         extra_context_to_return: Optional[Dict[str, Any]],
         resource_or_bundle: Dict[str, Any],
         separate_bundle_resources: bool,
-        text: str,
         total_count: int,
         url: Optional[str],
     ) -> Tuple[str, int]:
@@ -470,14 +468,14 @@ async def expand_or_separate_bundle_async(
             )
             resources_json = json.dumps(resource_separator_result.resources_dicts)
             total_count = resource_separator_result.total_count
-        elif resources:
+        elif len(resources) > 0:
             total_count = len(resources)
             if len(resources) == 1:
                 resources_json = json.dumps(resources[0])
             else:
                 resources_json = json.dumps(resources)
         else:
-            resources_json = text
+            resources_json = json.dumps(resources)
 
         return resources_json, total_count
 
@@ -619,7 +617,6 @@ def get_chunk_iterator() -> (
                                         extra_context_to_return=extra_context_to_return,
                                         resource_or_bundle=completed_resource,
                                         separate_bundle_resources=separate_bundle_resources,
-                                        text=json.dumps(completed_resource),
                                         total_count=total_count,
                                         url=url,
                                     )

helix_fhir_client_sdk/responses/fhir_update_response.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, AsyncGenerator
 
 
 class FhirUpdateResponse:
@@ -29,3 +29,31 @@ def __init__(
         self.access_token: Optional[str] = access_token
         self.status: int = status
         self.resource_type: Optional[str] = resource_type
+
+    def append(self, other: Optional["FhirUpdateResponse"]) -> None:
+        """
+        Appends another FhirUpdateResponse to this one
+
+        :param other: FhirUpdateResponse to append
+        """
+        if other:
+            self.responses += other.responses
+            self.error = (self.error or "") + (other.error or "")
+
+    @classmethod
+    async def from_async_generator(
+        cls, generator: AsyncGenerator["FhirUpdateResponse", None]
+    ) -> Optional["FhirUpdateResponse"]:
+        """
+        Reads a generator of FhirUpdateResponse and returns a single FhirUpdateResponse by appending all the FhirUpdateResponse
+
+        :param generator: generator of FhirUpdateResponse items
+        :return: FhirUpdateResponse
+        """
+        result: FhirUpdateResponse | None = None
+        async for value in generator:
+            if not result:
+                result = value
+            else:
+                result.append(value)
+        return result

helix_fhir_client_sdk/responses/get_result.py

@@ -1,9 +1,39 @@
 import dataclasses
-from typing import List, Dict, Any, Optional
+from typing import List, Dict, Any, Optional, AsyncGenerator
 
 
 @dataclasses.dataclass
 class GetResult:
     request_id: Optional[str]
     resources: List[Dict[str, Any]]
     response_headers: Optional[List[str]]
+
+    def append(self, other: Optional["GetResult"]) -> None:
+        """
+        Appends another GetResult to this one
+
+        :param other: GetResult to append
+        """
+        if other:
+            self.resources.extend(other.resources)
+            self.response_headers = (self.response_headers or []) + (
+                other.response_headers or []
+            )
+
+    @classmethod
+    async def from_async_generator(
+        cls, generator: AsyncGenerator["GetResult", None]
+    ) -> Optional["GetResult"]:
+        """
+        Reads a generator of GetResult and returns a single GetResult by appending all the GetResult
+
+        :param generator: generator of GetResult items
+        :return: GetResult
+        """
+        result: GetResult | None = None
+        async for value in generator:
+            if not result:
+                result = value
+            else:
+                result.append(value)
+        return result

helix_fhir_client_sdk/responses/paging_result.py

@@ -1,5 +1,5 @@
 import dataclasses
-from typing import List, Dict, Any, Optional
+from typing import List, Dict, Any, Optional, AsyncGenerator
 
 
 @dataclasses.dataclass
@@ -8,3 +8,34 @@ class PagingResult:
     resources: List[Dict[str, Any]]
     page_number: int
     response_headers: Optional[List[str]]
+
+    def append(self, other: Optional["PagingResult"]) -> None:
+        """
+        Appends another PagingResult to this one
+
+        :param other: PagingResult to append
+        """
+        if other:
+            self.resources.extend(other.resources)
+            self.response_headers = (self.response_headers or []) + (
+                other.response_headers or []
+            )
+            self.page_number = other.page_number
+
+    @classmethod
+    async def from_async_generator(
+        cls, generator: AsyncGenerator["PagingResult", None]
+    ) -> Optional["PagingResult"]:
+        """
+        Reads a generator of PagingResult and returns a single PagingResult by appending all the PagingResult
+
+        :param generator: generator of PagingResult items
+        :return: PagingResult
+        """
+        result: PagingResult | None = None
+        async for value in generator:
+            if not result:
+                result = value
+            else:
+                result.append(value)
+        return result