[issue-305, refactor] add methods to parse required/ optional fields …

…with exception handling Signed-off-by: Meret Behrens <meret.behrens@tngtech.com>
spdx · meretp · Dec 28, 2022 · Nov 25, 2022 · Dec 14, 2022 · Dec 14, 2022
commit 190cd5a4d81f48cb02ef9a97700bbcca0ff0629b
diff --git a/src/parser/json/annotation_parser.py b/src/parser/json/annotation_parser.py
@@ -15,7 +15,8 @@
 from src.model.annotation import Annotation, AnnotationType
 from src.parser.error import SPDXParsingError
 from src.parser.json.actor_parser import ActorParser
-from src.parser.json.dict_parsing_functions import datetime_from_str, try_construction_raise_parsing_error
+from src.parser.json.dict_parsing_functions import datetime_from_str, try_construction_raise_parsing_error, \
+    try_parse_optional_field_append_logger_when_failing, try_parse_required_field_append_logger_when_failing
 from src.parser.logger import Logger
 
 
@@ -78,21 +79,21 @@ def parse_annotations(self, annotations_dict_list: List[Dict], spdx_id: Optional
     def parse_annotation(self, annotation: Dict, spdx_id: Optional[str] = None) -> Annotation:
         logger = Logger()
         spdx_id: str = annotation.get("SPDXID") or spdx_id
-        try:
-            annotation_type: Optional[AnnotationType] = self.parse_annotation_type(annotation.get("annotationType"))
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            annotation_type = None
-        try:
-            annotator: Optional[Actor] = self.actor_parser.parse_actor(annotation.get("annotator"))
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            annotator = None
-        try:
-            annotation_date: Optional[datetime] = datetime_from_str(annotation.get("annotationDate"))
-        except TypeError:
-            logger.append("ValueError while parsing annotationDate.")
-            annotation_date = None
+
+        annotation_type: Optional[AnnotationType] = try_parse_required_field_append_logger_when_failing(
+            logger=logger, field=annotation.get("annotationType"),
+            method_to_parse=self.parse_annotation_type)
+
+        annotator: Optional[Actor] = try_parse_required_field_append_logger_when_failing(logger=logger,
+                                                                                         field=annotation.get(
+                                                                                             "annotator"),
+                                                                                         method_to_parse=self.actor_parser.parse_actor)
+
+        annotation_date: Optional[datetime] = try_parse_required_field_append_logger_when_failing(logger=logger,
+                                                                                                  field=annotation.get(
+                                                                                                      "annotationDate"),
+                                                                                                  method_to_parse=datetime_from_str)
+
         annotation_comment: str = annotation.get("comment")
         if logger.has_messages():
             raise SPDXParsingError([f"Error while parsing annotation: {logger.get_messages()}"])
@@ -112,16 +113,17 @@ def parse_annotation_type(annotation_type: str) -> AnnotationType:
 
     def parse_review(self, review_dict: Dict, spdx_id: str) -> Annotation:
         logger = Logger()
-        try:
-            annotator: Optional[Actor] = self.actor_parser.parse_actor(review_dict.get("reviewer"))
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            annotator = None
-        try:
-            annotation_date: Optional[datetime] = datetime_from_str(review_dict.get("reviewDate"))
-        except TypeError:
-            logger.append("ValueError while parsing reviewDate.")
-            annotation_date = None
+
+        annotator: Optional[Actor] = try_parse_optional_field_append_logger_when_failing(logger=logger,
+                                                                                         field=review_dict.get(
+                                                                                             "reviewer"),
+                                                                                         method_to_parse=self.actor_parser.parse_actor)
+
+        annotation_date: Optional[datetime] = try_parse_required_field_append_logger_when_failing(logger=logger,
+                                                                                                  field=review_dict.get(
+                                                                                                      "reviewDate"),
+                                                                                                  method_to_parse=datetime_from_str)
+
         annotation_type = AnnotationType.REVIEW
         comment: str = review_dict.get("comment")
         if logger.has_messages():

diff --git a/src/parser/json/checksum_parser.py b/src/parser/json/checksum_parser.py
@@ -11,7 +11,6 @@
 from typing import Dict, List
 
 from src.model.checksum import Checksum, ChecksumAlgorithm
-from src.model.typing.constructor_type_errors import ConstructorTypeErrors
 from src.parser.error import SPDXParsingError
 from src.parser.json.dict_parsing_functions import transform_json_str_to_enum_name, try_construction_raise_parsing_error
 from src.parser.logger import Logger

diff --git a/src/parser/json/creation_info_parser.py b/src/parser/json/creation_info_parser.py
@@ -20,8 +20,9 @@
 from src.parser.error import SPDXParsingError
 from src.parser.json.actor_parser import ActorParser
 from src.parser.json.checksum_parser import ChecksumParser
-from src.parser.json.dict_parsing_functions import datetime_from_str, parse_optional_field, \
-    try_construction_raise_parsing_error
+from src.parser.json.dict_parsing_functions import datetime_from_str, try_construction_raise_parsing_error, \
+    try_parse_optional_field_append_logger_when_failing, \
+    try_parse_required_field_append_logger_when_failing
 from src.parser.logger import Logger
 
 
@@ -47,33 +48,28 @@ def parse_creation_info(self, doc_dict: Dict) -> CreationInfo:
         if creation_info_dict is None:
             logger.append("CreationInfo is not valid.")
             raise SPDXParsingError([f"Error while parsing doc {name}: {logger.get_messages()}"])
-        try:
-            list_of_creators: List[str] = creation_info_dict.get("creators")
-            creators: List[Actor] = self.parse_creators(list_of_creators)
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            creators = []
-        try:
-            created: Optional[datetime] = datetime_from_str(creation_info_dict.get("created"))
-        except ValueError:
-            logger.append("Error while parsing created")
-            created = None
+
+        list_of_creators: List[str] = creation_info_dict.get("creators")
+        creators: List[Actor] = try_parse_required_field_append_logger_when_failing(logger=logger,
+                                                                                    field=list_of_creators,
+                                                                                    method_to_parse=self.parse_creators,
+                                                                                    default=[])
+
+        created: Optional[datetime] = try_parse_required_field_append_logger_when_failing(logger=logger,
+                                                                                          field=creation_info_dict.get(
+                                                                                              "created"),
+                                                                                          method_to_parse=datetime_from_str)
 
         creator_comment: Optional[str] = creation_info_dict.get("comment")
         data_license: str = doc_dict.get("dataLicense")
-        try:
-            external_document_refs: List[ExternalDocumentRef] = parse_optional_field(
-                doc_dict.get("externalDocumentRefs"),
-                self.parse_external_document_refs)
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            external_document_refs = []
-        try:
-            license_list_version: Optional[Version] = parse_optional_field(creation_info_dict.get("licenseListVersion"),
-                                                                           self.parse_version)
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            license_list_version = None
+
+        external_document_refs: List[ExternalDocumentRef] = try_parse_optional_field_append_logger_when_failing(
+            logger=logger, field=doc_dict.get("externalDocumentRefs"),
+            method_to_parse=self.parse_external_document_refs)
+        license_list_version: Optional[Version] = try_parse_optional_field_append_logger_when_failing(logger=logger,
+                                                                                                      field=creation_info_dict.get(
+                                                                                                          "licenseListVersion"),
+                                                                                                      method_to_parse=self.parse_version)
         document_comment: Optional[str] = doc_dict.get("comment")
         if logger.has_messages():
             raise SPDXParsingError([f"Error while parsing doc {name}: {logger.get_messages()}"])
@@ -114,22 +110,23 @@ def parse_external_document_refs(self, external_document_refs_dict: List[Dict])
         logger = Logger()
         external_document_refs = []
         for external_ref_dict in external_document_refs_dict:
-            try:
-                external_doc_ref: ExternalDocumentRef = self.parse_external_doc_ref(external_ref_dict)
-                external_document_refs.append(external_doc_ref)
-            except SPDXParsingError as err:
-                logger.append_all(err.get_messages())
+            external_doc_ref: ExternalDocumentRef = try_parse_optional_field_append_logger_when_failing(logger=logger,
+                                                                                                        field=external_ref_dict,
+                                                                                                        method_to_parse=self.parse_external_doc_ref)
+
+            external_document_refs.append(external_doc_ref)
+
         if logger.has_messages():
             raise SPDXParsingError(logger.get_messages())
         return external_document_refs
 
     def parse_external_doc_ref(self, external_doc_ref_dict: Dict) -> ExternalDocumentRef:
         logger = Logger()
-        try:
-            checksum: Optional[Checksum] = self.checksum_parser.parse_checksum(external_doc_ref_dict.get("checksum"))
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            checksum = None
+        checksum: Optional[Checksum] = try_parse_required_field_append_logger_when_failing(logger=logger,
+                                                                                           field=external_doc_ref_dict.get(
+                                                                                               "checksum"),
+                                  
A36C
                                                         method_to_parse=self.checksum_parser.parse_checksum)
+
         external_document_id: str = external_doc_ref_dict.get("externalDocumentId")
         spdx_document: str = external_doc_ref_dict.get("spdxDocument")
         if logger.has_messages():

diff --git a/src/parser/json/dict_parsing_functions.py b/src/parser/json/dict_parsing_functions.py
@@ -1,24 +1,26 @@
 from datetime import datetime
 from typing import Any, Callable, Dict
 
-from src.model.external_document_ref import ExternalDocumentRef
 from src.model.typing.constructor_type_errors import ConstructorTypeErrors
 from src.parser.error import SPDXParsingError
+from src.parser.logger import Logger
 
 
-def parse_optional_field(field: Any, method_to_parse:Callable=lambda x: x, default=None):
+def parse_optional_field(field: Any, method_to_parse: Callable = lambda x: x, default=None):
     if not field:
         return default
     return method_to_parse(field)
 
 
-def datetime_from_str(created: str) -> datetime:
-    date = datetime.strptime(created, "%Y-%m-%dT%H:%M:%SZ")
+def datetime_from_str(date_str: str) -> datetime:
+    if not isinstance(date_str, str):
+        raise SPDXParsingError([f"Could not convert str to datetime, invalid type: {type(date_str).__name__}"])
+    date = datetime.strptime(date_str, "%Y-%m-%dT%H:%M:%SZ")
     return date
 
 
 def transform_json_str_to_enum_name(json_str: str) -> str:
-    return json_str.replace("-","_").upper()
+    return json_str.replace("-", "_").upper()
 
 
 def try_construction_raise_parsing_error(object_to_construct: Any, args_for_construction: Dict) -> Any:
@@ -27,3 +29,21 @@ def try_construction_raise_parsing_error(object_to_construct: Any, args_for_cons
     except ConstructorTypeErrors as err:
         raise SPDXParsingError([f"Error while constructing {object_to_construct.__name__}: {err.get_messages()}"])
     return constructed_object
+
+
+def try_parse_optional_field_append_logger_when_failing(logger: Logger, field: Any, method_to_parse: Callable, default=None):
+    try:
+        parsed_element = parse_optional_field(field=field, method_to_parse=method_to_parse, default=default)
+    except SPDXParsingError as err:
+        logger.append_all(err.get_messages())
+        parsed_element = default
+    return parsed_element
+
+def try_parse_required_field_append_logger_when_failing(logger: Logger, field: Any, method_to_parse: Callable, default=None):
+    try:
+        parsed_element = method_to_parse(field)
+    except SPDXParsingError as err:
+        logger.append_all(err.get_messages())
+        parsed_element = default
+    return parsed_element
+
diff --git a/src/parser/json/file_parser.py b/src/parser/json/file_parser.py
@@ -17,7 +17,8 @@
 from src.model.spdx_none import SpdxNone
 from src.parser.error import SPDXParsingError
 from src.parser.json.checksum_parser import ChecksumParser
-from src.parser.json.dict_parsing_functions import parse_optional_field, try_construction_raise_parsing_error
+from src.parser.json.dict_parsing_functions import try_construction_raise_parsing_error, \
+    try_parse_optional_field_append_logger_when_failing, try_parse_required_field_append_logger_when_failing
 from src.parser.json.license_expression_parser import LicenseExpressionParser
 from src.parser.logger import Logger
 
@@ -50,37 +51,34 @@ def parse_file(self, file_dict: Dict) -> Optional[File]:
         name: str = file_dict.get("fileName")
         spdx_id: str = file_dict.get("SPDXID")
         checksums_list: List[Dict] = file_dict.get("checksums")
-        try:
-            checksums: List[Checksum] = self.checksum_parser.parse_checksums(checksums_list)
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            checksums = []
+
+        checksums: List[Checksum] = try_parse_required_field_append_logger_when_failing(logger=logger,
+                                                                                        field=checksums_list,
+                                                                                        method_to_parse=self.checksum_parser.parse_checksums)
 
         attribution_texts: Optional[str] = file_dict.get("attributionTexts")
         comment: Optional[str] = file_dict.get("comment")
         copyright_text: Optional[str] = file_dict.get("copyrightText")
         file_contributors: List[str] = file_dict.get("fileContributors")
-        try:
-            file_types: List[FileType] = parse_optional_field(file_dict.get("fileTypes"), self.parse_file_types)
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            file_types = []
+        file_types: List[FileType] = try_parse_optional_field_append_logger_when_failing(logger=logger,
+                                                                                         field=file_dict.get(
+                                                                                             "fileTypes"),
+                                                                                         method_to_parse=self.parse_file_types)
+
         license_comments: Optional[str] = file_dict.get("licenseComments")
-        try:
-            license_concluded: Optional[Union[LicenseExpression, SpdxNoAssertion, SpdxNone]] = parse_optional_field(
-                file_dict.get("licenseConcluded"),
-                self.license_expression_parser.parse_license_expression)
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            license_concluded = None
-        try:
-            license_info_in_files: Optional[
-                Union[List[LicenseExpression], SpdxNoAssertion, SpdxNone]] = parse_optional_field(
-                file_dict.get("licenseInfoInFiles"),
-                self.license_expression_parser.parse_license_expression)
-        except SPDXParsingError as err:
-            logger.append_all(err.get_messages())
-            license_info_in_files = None
+
+        license_concluded: Optional[Union[
+            LicenseExpression, SpdxNoAssertion, SpdxNone]] = try_parse_optional_field_append_logger_when_failing(
+            logger=logger, field=file_dict.get("licenseConcluded"),
+            method_to_parse=self.license_expression_parser.parse_license_expression)
+
+        license_info_in_files: Optional[
+            Union[List[
+                LicenseExpression], SpdxNoAssertion, SpdxNone]] = try_parse_optional_field_append_logger_when_failing(
+            logger=logger,
+            field=file_dict.get("licenseInfoInFiles"),
+            method_to_parse=self.license_expression_parser.parse_license_expression)
+
         notice_text: Optional[str] = file_dict.get("noticeText")
 
         if logger.has_messages():