Revert "import fhirpy_types_r4b"

This reverts commit bbb791f.

Author: alexlipovka

Pipfile

@@ -31,7 +31,6 @@ aidbox-python-sdk = ">=0.1.16"
 fhirpathpy = ">=2.0.3"
 sqlalchemy = ">=1.4.44"
 fhirpy = "~=2.2.0"
-fhirpy-types-r4b = {git = "https://github.com/beda-software/fhir-py-types-r4b.git"}
 aiohttp-cors = "~=0.7.0"
 simplejson = "*"
 fpml = ">=0.2.0"

Pipfile.lock

@@ -1,4 +1,3 @@
-from fhirpy_types_r4b import Questionnaire
 from funcy.seqs import flatten
 
 from .getters import (
@@ -71,7 +70,7 @@ def _merge_root_ext(accumulated: list, ext: dict):
         accumulated.append(ext)
 
 
-async def assemble(client, questionnaire: Questionnaire):
+async def assemble(client, questionnaire):
     original_exts = list(questionnaire.get("extension", []))
     non_whitelisted_root_exts = [
         e for e in original_exts if e.get("url") not in WHITELISTED_EXTENSION_URLS
@@ -170,7 +169,7 @@ def _load_sub_questionnaire(sub_exts, root_contained, parent_item, item, sub_que
     return item
 
 
-def _propagate_extensions(parent_item, sub_questionnaire: Questionnaire):
+def _propagate_extensions(parent_item, sub_questionnaire):
     parent_exts = list(parent_item.get("extension", []))
     for ext in sub_questionnaire.get("extension", []):
         if ext.get("url") in PROPAGATE_EXTENSION_URLS:
@@ -212,7 +211,7 @@ def _assemble_questionnaire(
     return resp
 
 
-def _validate_assemble_context(questionnaire: Questionnaire, variables: dict):
+def _validate_assemble_context(questionnaire, variables: dict):
     assemble_ctx = get_assemble_context(questionnaire.get("extension", []))
     if not assemble_ctx:
         return False

app/sdc/assemble.py

@@ -1,13 +1,11 @@
-from fhirpy_types_r4b import Questionnaire, QuestionnaireItem
-
 from app.cached_fhirpath import fhirpath
 
 from .exception import ConstraintCheckOperationOutcome
 from .getters import get_item_constraints, get_launch_context
 from .utils import load_source_queries, validate_context
 
 
-async def constraint_check(client, questionnaire: Questionnaire, env, *, legacy_behavior=False):
+async def constraint_check(client, questionnaire, env, *, legacy_behavior=False):
     launch_context = get_launch_context(questionnaire.get("extension", []))
     if launch_context:
         validate_context(launch_context, env)
@@ -19,9 +17,7 @@ async def constraint_check(client, questionnaire: Questionnaire, env, *, legacy_
     return env["QuestionnaireResponse"]
 
 
-def _constraint_check_for_item(
-    errors, questionnaire_item: Questionnaire | QuestionnaireItem, env, *, legacy_behavior=False
-):
+def _constraint_check_for_item(errors, questionnaire_item, env, *, legacy_behavior=False):
     for constraint in get_item_constraints(questionnaire_item.get("extension", [])):
         expression = constraint["expression"]
         result = fhirpath({}, expression, env)

app/sdc/constraint_check.py

@@ -1,8 +1,6 @@
-from fhirpy_types_r4b import Questionnaire
-
 from .utils import load_source_queries
 
 
-async def get_questionnaire_context(client, questionnaire: Questionnaire, env):
+async def get_questionnaire_context(client, questionnaire, env):
     await load_source_queries(client, questionnaire, env)
     return env

app/sdc/context.py

@@ -1,6 +1,5 @@
 import logging
 
-from fhirpy_types_r4b import Questionnaire
 from funcy import is_list
 
 from .getters import (
@@ -22,7 +21,7 @@
 logger = logging.getLogger(__name__)
 
 
-async def populate(client, fhir_questionnaire: Questionnaire, env, *, sdc_api: bool = False):
+async def populate(client, fhir_questionnaire, env, *, sdc_api: bool = False):
     exts = fhir_questionnaire.get("extension", [])
     launch_context = get_launch_context(exts)
     if launch_context:

app/sdc/populate.py

@@ -6,7 +6,6 @@
 from fhirpy import AsyncFHIRClient
 from fhirpy.base.exceptions import OperationOutcome
 from fhirpy.base.utils import get_by_path
-from fhirpy_types_r4b import Parameters, Questionnaire
 from fpml import resolve_template
 from funcy.seqs import first
 from funcy.strings import re_all
@@ -112,7 +111,7 @@ def _update_link_id_or_question(value, key):
     return _update_link_id_or_question
 
 
-def prepare_link_ids(questionnaire: Questionnaire, variables):
+def prepare_link_ids(questionnaire, variables):
     return walk_dict(questionnaire, update_link_id_or_question(variables))
 
 
@@ -159,15 +158,15 @@ def prepare_assemble_variables(item):
     return variables
 
 
-def is_sdc_api(parameters: Parameters | None) -> bool:
+def is_sdc_api(parameters: dict | None) -> bool:
     """True when Parameters use SDC `$populate` input shape (`context` and/or `subject`)."""
     if not parameters or parameters.get("resourceType") != "Parameters":
         return False
     return any(p.get("name") in ("context", "subject") for p in parameters.get("parameter", []))
 
 
 async def parameter_to_env(
-    client: AsyncFHIRClient, resource: Parameters, is_fhir: bool = True
+    client: AsyncFHIRClient, resource, is_fhir: bool = True
 ) -> dict[str, Any]:
     # TODO: add support for repeating values (with same name)
     env: dict[str, Any] = {}
@@ -213,7 +212,7 @@ def parse_parameter_value(parameter, is_fhir: bool) -> tuple[Any, str]:
     return value[polimorphic_key] if polimorphic_key else None, polimorphic_key
 
 
-def get_external_fhir_base_url_from_resource(resource: Parameters | None, is_fhir: bool):
+def get_external_fhir_base_url_from_resource(resource: dict | None, is_fhir: bool):
     if not resource or resource.get("resourceType") != "Parameters":
         return None
     for param in resource.get("parameter", []):
@@ -226,7 +225,7 @@ def get_external_fhir_base_url_from_resource(resource: Parameters | None, is_fhi
     return None
 
 
-async def load_source_queries(client, questionnaire: Questionnaire, env):
+async def load_source_queries(client, questionnaire, env):
     contained = {item["id"]: item for item in questionnaire.get("contained", [])}
     source_queries = get_source_queries(questionnaire.get("extension", []))