Merge pull request #415 from bento-platform/fix/type-hinting

fix: extension schema dict type hinting & export decl

chord_metadata_service/chord/ingest/phenopackets.py

@@ -11,6 +11,7 @@
 from chord_metadata_service.phenopackets.schemas import PHENOPACKET_SCHEMA
 from chord_metadata_service.patients.values import KaryotypicSex
 from chord_metadata_service.restapi.schema_utils import patch_project_schemas
+from chord_metadata_service.restapi.types import ExtensionSchemaDict
 from chord_metadata_service.restapi.utils import iso_duration_to_years
 
 from .exceptions import IngestError
@@ -19,7 +20,7 @@
 from .schema import schema_validation
 from .utils import get_output_or_raise, map_if_list, query_and_check_nulls, workflow_file_output_to_path
 
-from typing import Any, Optional, Union
+from typing import Any, Dict, Iterable, Optional, Union
 
 
 def get_or_create_phenotypic_feature(pf: dict) -> pm.PhenotypicFeature:
@@ -310,14 +311,17 @@ def ingest_phenopacket_workflow(workflow_outputs, table_id) -> Union[list[pm.Phe
             json_data = json.load(jf)
 
     project_id = Project.objects.get(datasets__table_ownership=table_id)
-    project_schemas = ProjectJsonSchema.objects.filter(project_id=project_id).values(
+    project_schemas: Iterable[ExtensionSchemaDict] = ProjectJsonSchema.objects.filter(project_id=project_id).values(
         "json_schema",
         "required",
         "schema_type",
     )
 
     # Map with key:schema_type and value:json_schema
-    extension_schemas = {proj_schema["schema_type"].lower(): proj_schema for proj_schema in project_schemas}
+    extension_schemas: Dict[str, ExtensionSchemaDict] = {
+        proj_schema["schema_type"].lower(): proj_schema
+        for proj_schema in project_schemas
+    }
     json_schema = patch_project_schemas(PHENOPACKET_SCHEMA, extension_schemas)
 
     # First, validate all phenopackets

chord_metadata_service/restapi/schema_utils.py

@@ -1,8 +1,10 @@
-from chord_metadata_service.logger import logger
 from bento_lib.search import queries as q
-from .description_utils import describe_schema
-from typing import List, Optional, Dict
+from chord_metadata_service.logger import logger
 from copy import deepcopy
+from typing import List, Optional, Dict
+
+from .description_utils import describe_schema
+from .types import ExtensionSchemaDict
 
 __all__ = [
     "merge_schema_dictionaries",
@@ -13,6 +15,7 @@
     "tag_ids_and_describe",
     "customize_schema",
     "schema_list",
+    "patch_project_schemas",
 ]
 
 
@@ -151,7 +154,7 @@ def schema_list(schema):
     }
 
 
-def patch_project_schemas(base_schema: dict, extension_schemas: Dict[str, object]) -> dict:
+def patch_project_schemas(base_schema: dict, extension_schemas: Dict[str, ExtensionSchemaDict]) -> dict:
     if not isinstance(base_schema, dict) or "type" not in base_schema:
         return base_schema
 

chord_metadata_service/restapi/types.py

@@ -0,0 +1,11 @@
+from typing import TypedDict
+
+__all__ = [
+    "ExtensionSchemaDict",
+]
+
+
+class ExtensionSchemaDict(TypedDict):
+    json_schema: dict
+    required: bool
+    schema_type: str