simply and reformat

staging_service/import_specifications/individual_parsers.py

@@ -2,6 +2,7 @@
 Contains parser functions for use with the file parser framework.
 """
 
+from collections import defaultdict
 import csv
 import json
 import math
@@ -63,9 +64,6 @@
 _DTS_INSTRUCTIONS_KEY = "instructions"
 _DTS_INSTRUCTIONS_DATATYPE_KEY = "data_type"
 _DTS_INSTRUCTIONS_PARAMETERS_KEY = "parameters"
-_DTS_INSTRUCTIONS_REQUIRED_KEYS = [_DTS_INSTRUCTIONS_DATATYPE_KEY, _DTS_INSTRUCTIONS_PARAMETERS_KEY]
-_DTS_INSTRUCTIONS_PROTOCOL_KEY = "protocol"
-_DTS_INSTRUCTIONS_PROTOCOL = "KBase narrative import"
 _DTS_INSTRUCTIONS_OBJECTS_KEY = "objects"
 
 
@@ -359,62 +357,35 @@ def parse_dts_manifest(path: Path, validator: Draft202012Validator) -> ParseResu
     results = {}
     try:
         with open(path, "r") as manifest:
-            manifest_json = json.load(manifest)
-        if not isinstance(manifest_json, dict):
+            manifest = json.load(manifest)
+        if not isinstance(manifest, dict):
             return _error(Error(ErrorType.PARSE_FAIL, "Manifest is not a dictionary", spcsrc))
-        for err in validator.iter_errors(manifest_json):
+        for err in validator.iter_errors(manifest):
             err_str = err.message
             err_path = list(err.absolute_path)
             if err_path:
                 err_str += f" at {err_path}"
             errors.append(Error(ErrorType.PARSE_FAIL, err_str, spcsrc))
         if not errors:
-            results = _process_dts_manifest(manifest_json, spcsrc)
-
+            results = defaultdict(list)
+            instructions = manifest[_DTS_INSTRUCTIONS_KEY]
+            for resource_obj in instructions[_DTS_INSTRUCTIONS_OBJECTS_KEY]:
+                datatype = resource_obj[_DTS_INSTRUCTIONS_DATATYPE_KEY]
+                parameters = frozendict(resource_obj[_DTS_INSTRUCTIONS_PARAMETERS_KEY])
+                results[datatype].append(parameters)
+            # Re-package results as a dict of {datatype: ParseResult}
+            results = {
+                source: ParseResult(spcsrc, tuple(parsed)) for source, parsed in results.items()
+            }
     except json.JSONDecodeError:
         return _error(Error(ErrorType.PARSE_FAIL, "File must be in JSON format", spcsrc))
     except FileNotFoundError:
         return _error(Error(ErrorType.FILE_NOT_FOUND, source_1=spcsrc))
     except IsADirectoryError:
         return _error(Error(ErrorType.PARSE_FAIL, "The given path is a directory", spcsrc))
-    except _ParseException as err:
-        return _error(err.args[0])
     if errors:
         return ParseResults(errors=tuple(errors))
     elif results:
         return ParseResults(frozendict(results))
     else:
         return _error(Error(ErrorType.PARSE_FAIL, "No import specification data in file", spcsrc))
-
-
-def _process_dts_manifest(
-    manifest: dict[str, Any], spcsrc: SpecificationSource
-) -> Tuple[dict[str, ParseResult]]:
-    """Parse the DTS manifest file and return the results and a list of errors if applicable.
-
-    Results are returned as a dictionary where keys are data types, and values are ParseResults for that data type.
-    This assumes that the manifest has the correct structure, i.e. is validated via jsonschema.
-    Will raise KeyErrors otherwise.
-    """
-    results = {}
-    instructions = manifest[_DTS_INSTRUCTIONS_KEY]
-    # Make sure the protocol value matches.
-    if instructions[_DTS_INSTRUCTIONS_PROTOCOL_KEY] != _DTS_INSTRUCTIONS_PROTOCOL:
-        raise _ParseException(
-            Error(
-                ErrorType.PARSE_FAIL,
-                f"The instructions protocol must be '{_DTS_INSTRUCTIONS_PROTOCOL}'",
-                spcsrc,
-            )
-        )
-    for resource_obj in instructions[_DTS_INSTRUCTIONS_OBJECTS_KEY]:
-        datatype = resource_obj[_DTS_INSTRUCTIONS_DATATYPE_KEY]
-        parameters = frozendict(resource_obj[_DTS_INSTRUCTIONS_PARAMETERS_KEY])
-        if datatype not in results:
-            results[datatype] = []
-        results[datatype].append(parameters)
-    # Package results as a dict of {datatype: ParseResult}
-    parsed_result = {
-        source: ParseResult(spcsrc, tuple(parsed)) for source, parsed in results.items()
-    }
-    return parsed_result

tests/import_specifications/test_individual_parsers.py

@@ -17,14 +17,14 @@
     parse_dts_manifest,
     parse_excel,
     parse_tsv,
-    _DTS_INSTRUCTIONS_PROTOCOL,
 )
 from tests.test_app import FileUtil
 from tests.test_utils import bootstrap_config
 from jsonschema import Draft202012Validator
 
 
 _TEST_DATA_DIR = (Path(__file__).parent / "test_data").resolve()
+_DTS_INSTRUCTIONS_PROTOCOL = "KBase narrative import"
 
 
 @pytest.fixture(scope="module", name="temp_dir")