#20 - change CoreOrExtType to enum for type hinting (review feedback)

Author: patkyn

src/dwcahandler/dwca/__init__.py

@@ -25,17 +25,13 @@
 from enum import Enum
 from functools import wraps
 from typing import Optional, Union
-
+import logging
 import pandas as pd
 
 
 class CoreOrExtType(Enum):
     CORE = "core"
     EXTENSION = "extension"
-@dataclass(frozen=True)
-class MetaDefaultFields:
-    ID: str = "id"
-    CORE_ID: str = "coreid"
 
 # Default keys for content when creating dwca
 DefaultKeys = namedtuple("DefaultKeys", ["EVENT", "OCCURRENCE"])(
@@ -184,8 +180,17 @@ class Defaults:
     # Translation csv encoding values
     translate_table: dict = field(init=False,
                                   default_factory=lambda: {'LF': '\r\n', '\\t': '\t', '\\n': '\n'})
+    MetaDefaultFields: namedtuple = namedtuple("MetaDefaultFields", ["ID", "CORE_ID"])(
+                                        ID="id",
+                                        CORE_ID="coreid"
+                                    )
+
 
 
+# Imports at end of file to allow classes to be used
+from dwcahandler.dwca.terms import Terms, NsPrefix
+from dwcahandler.dwca.dwca_meta import (MetaElementTypes, MetaElementInfo, MetaDwCA,
+                                        MetaElementAttributes, get_meta_class_row_type)
 @dataclass
 class ContentData:
     """A class describing the content data used for core and extension.
@@ -220,13 +225,8 @@ def add_data(self, other_csv_file_type: ContentData):
             self.type = other_csv_file_type.type
         return False
 
-
+from dwcahandler.dwca.eml import Eml
 from dwcahandler.dwca.base_dwca import BaseDwca
-from dwcahandler.dwca.core_dwca import DfContent, Dwca
+from dwcahandler.dwca.core_dwca import Dwca, DfContent
 from dwcahandler.dwca.dwca_factory import DwcaHandler
-from dwcahandler.dwca.dwca_meta import (MetaDwCA, MetaElementAttributes,
-                                        MetaElementInfo, MetaElementTypes,
-                                        get_meta_class_row_type)
-from dwcahandler.dwca.eml import Eml
-# Imports at end of file to allow classes to be used
-from dwcahandler.dwca.terms import NsPrefix, Terms
+

src/dwcahandler/dwca/core_dwca.py

@@ -21,7 +21,7 @@
 from numpy import nan
 from pandas.errors import EmptyDataError
 from pandas.io import parsers
-from dwcahandler.dwca import (BaseDwca, CoreOrExtType, MetaDefaultFields, CSVEncoding,
+from dwcahandler.dwca import (BaseDwca, CoreOrExtType, CSVEncoding,
                               ContentData, Defaults, Eml, Terms, get_keys,
                               MetaDwCA, MetaElementInfo, MetaElementTypes,
                               MetaElementAttributes, Stat, record_diff_stat)
@@ -93,9 +93,9 @@ def _update_core_ids(self, core_df) -> str:
         :param core_df: The data frame to generate identifiers for
         return id field
         """
-        if MetaDefaultFields.ID not in core_df.columns.to_list():
-            core_df.insert(0, MetaDefaultFields.ID, core_df.apply(lambda _: uuid.uuid4(), axis=1), False)
-            return MetaDefaultFields.ID
+        if self.defaults_prop.MetaDefaultFields.ID not in core_df.columns.to_list():
+            core_df.insert(0, self.defaults_prop.MetaDefaultFields.ID, core_df.apply(lambda _: uuid.uuid4(), axis=1), False)
+            return self.defaults_prop.MetaDefaultFields.ID
         else:
             raise ValueError("core df should not contain id column")
 
@@ -140,23 +140,23 @@ def _update_extension_ids(self, csv_content: pd.DataFrame, core_df_content: pd.D
                 set(link_col).issubset(set(csv_content.index.names))):
             csv_content.reset_index(inplace=True, drop=True)
 
-        csv_content = csv_content.merge(core_df_content.loc[:, MetaDefaultFields.ID],
+        csv_content = csv_content.merge(core_df_content.loc[:, self.defaults_prop.MetaDefaultFields.ID],
                                         left_on=link_col,
                                         right_on=link_col, how='outer')
 
-        if MetaDefaultFields.ID in csv_content.columns.to_list():
-            unmatched_content = csv_content[csv_content[MetaDefaultFields.ID].isnull()]
-            unmatched_content = unmatched_content.drop(columns=[MetaDefaultFields.ID])
+        if self.defaults_prop.MetaDefaultFields.ID in csv_content.columns.to_list():
+            unmatched_content = csv_content[csv_content[self.defaults_prop.MetaDefaultFields.ID].isnull()]
+            unmatched_content = unmatched_content.drop(columns=[self.defaults_prop.MetaDefaultFields.ID])
             if len(unmatched_content) > 0:
                 log.info("There are orphaned keys in extension file")
                 pd.set_option("display.max_columns", 7)
                 pd.set_option('display.max_colwidth', 15)
                 pd.set_option('display.max_rows', 10)
                 log.info("\n%s", unmatched_content)
-            csv_content = csv_content[~csv_content[MetaDefaultFields.ID].isnull()]
-            col = csv_content.pop(MetaDefaultFields.ID)
+            csv_content = csv_content[~csv_content[self.defaults_prop.MetaDefaultFields.ID].isnull()]
+            col = csv_content.pop(self.defaults_prop.MetaDefaultFields.ID)
             csv_content.insert(0, col.name, col)
-            csv_content.rename(columns={MetaDefaultFields.ID: ext_core_id_field}, inplace=True)
+            csv_content.rename(columns={self.defaults_prop.MetaDefaultFields.ID: ext_core_id_field}, inplace=True)
             return csv_content, ext_core_id_field
         else:
             raise ValueError("Something is not right. The core id failed to be created")
@@ -207,8 +207,9 @@ def _find_fields_with_zero_idx(meta_element_fields: list):
         def _add_first_id_field_if_exists(meta_element: MetaElementAttributes):
             zero_index_exist = _find_fields_with_zero_idx(meta_element.fields)
             if meta_element.core_id and meta_element.core_id.index and not zero_index_exist:
-                return [MetaDefaultFields.ID] if meta_element.meta_element_type.core_or_ext_type == CoreOrExtType.CORE \
-                    else [MetaDefaultFields.CORE_ID]
+                return [self.defaults_prop.MetaDefaultFields.ID] if (
+                        meta_element.meta_element_type.core_or_ext_type == CoreOrExtType.CORE) \
+                    else [self.defaults_prop.MetaDefaultFields.CORE_ID]
             else:
                 return []
 
@@ -287,7 +288,7 @@ def _update_values(self, df_content, delta_df_content, keys, stat):
         :return: The updated content
         """
         # Extract columns that need updating, excluding self.keys and id
-        non_update_column = list(MetaDefaultFields)
+        non_update_column = list(self.defaults_prop.MetaDefaultFields)
         non_update_column.extend(keys)
         update_columns = [i for i in delta_df_content.columns.to_list()
                           if i not in non_update_column]
@@ -429,12 +430,13 @@ def _extract_core_keys(self, core_content, keys):
         :return: A data frame indexed by the `id` column that contains the
                 key elements for each record
         """
-        columns = [MetaDefaultFields.ID] if MetaDefaultFields.ID in core_content.columns.tolist() else []
+        columns = [self.defaults_prop.MetaDefaultFields.ID] \
+            if self.defaults_prop.MetaDefaultFields.ID in core_content.columns.tolist() else []
         if all(key in core_content.columns for key in keys):
             columns.extend(keys)
             df = core_content[columns]
-            if MetaDefaultFields.ID in core_content.columns.tolist():
-                df.set_index(MetaDefaultFields.ID, drop=True, inplace=True)
+            if self.defaults_prop.MetaDefaultFields.ID in core_content.columns.tolist():
+                df.set_index(self.defaults_prop.MetaDefaultFields.ID, drop=True, inplace=True)
         else:
             raise ValueError(f"Keys does not exist in core content {''.join(keys)}")
         return df
@@ -869,17 +871,17 @@ def validate_content(self, content_to_validate: dict = None, error_file: str = N
 
                 if not self.check_duplicates(keys_df, content.keys, error_file):
                     log.error("Validation failed for %s %s content for duplicates keys %s",
-                              content.meta_info.core_or_ext_type, content.meta_info.type, content.keys)
+                              content.meta_info.core_or_ext_type.value, content.meta_info.type, content.keys)
                     validation_content_success = False
 
                 if not self._validate_columns(content):
                     log.error("Validation failed for %s %s content for duplicate columns",
-                              content.meta_info.core_or_ext_type, content.meta_info.type)
+                              content.meta_info.core_or_ext_type.value, content.meta_info.type)
                     validation_content_success = False
 
                 if validation_content_success:
                     log.info("Validation successful for %s %s content for unique keys %s",
-                             content.meta_info.core_or_ext_type, content.meta_info.type, content.keys)
+                             content.meta_info.core_or_ext_type.value, content.meta_info.type, content.keys)
                 else:
                     validation_success = False
 

src/dwcahandler/dwca/dwca_meta.py

@@ -11,7 +11,7 @@
 import re
 from dataclasses import dataclass, field
 from typing import Optional
-from dwcahandler.dwca import CSVEncoding, CoreOrExtType, Terms, MetaDefaultFields
+from dwcahandler.dwca import CSVEncoding, CoreOrExtType, Terms, Defaults
 from enum import Enum
 
 
@@ -72,7 +72,7 @@ class MetaElementAttributes:
 @dataclass
 class MetaDwCA:
     """Complete Metadata for a DwCA including dataset metadata and schema information"""
-    eml_xml_filename: str = field(default='eml.xml')
+    eml_xml_filename: str = field(default=Defaults.eml_xml_filename)
     dwca_meta: ET.Element = field(init=False)
     meta_elements: list[MetaElementAttributes] = field(default_factory=list, init=False)
 
@@ -89,10 +89,10 @@ def extract_field_attr_value(field_elm, attrib):
 
         fields = node_elm.findall(f'{ns}field')
         id_field = []
-        if core_or_ext_type == 'core':
-            id_field = node_elm.findall(f'{ns}id')
+        if core_or_ext_type == CoreOrExtType.CORE:
+            id_field = node_elm.findall(f'{ns}{Defaults.MetaDefaultFields.ID}')
         else:
-            id_field = node_elm.findall(f'{ns}coreid')
+            id_field = node_elm.findall(f'{ns}{Defaults.MetaDefaultFields.CORE_ID}')
         file_name = node_elm.find(f'{ns}files').find(f'{ns}location').text
         meta_element_info = MetaElementInfo(
             core_or_ext_type=core_or_ext_type,
@@ -138,11 +138,11 @@ def read_meta_file(self, meta_file):
         tree = ET.parse(meta_file)
         root = tree.getroot()
         ns = self._get_namespace(root)
-        node_elm = root.find(f'{ns}{CoreOrExtType.CORE}')
+        node_elm = root.find(f"{ns}{CoreOrExtType.CORE.value}")
         self.meta_elements = [self.__extract_meta_info(ns, node_elm, CoreOrExtType.CORE)]
         self.meta_elements.extend(
             [self.__extract_meta_info(ns, ne, CoreOrExtType.EXTENSION)
-             for ne in root.findall(f'{ns}{CoreOrExtType.EXTENSION}')])
+             for ne in root.findall(f"{ns}{CoreOrExtType.EXTENSION.value}")])
 
     def remove_meta_elements(self, exts_to_remove):
         """Remove extension files from the metadata
@@ -211,7 +211,7 @@ def _build_meta_xml(self, meta_elem_attrib: MetaElementAttributes):
 
         :param meta_elem_attrib: The meta information for the row
         """
-        elem = ET.SubElement(self.dwca_meta, meta_elem_attrib.meta_element_type.core_or_ext_type)
+        elem = ET.SubElement(self.dwca_meta, meta_elem_attrib.meta_element_type.core_or_ext_type.value)
         elem.attrib['encoding'] = meta_elem_attrib.meta_element_type.charset_encoding
         elem.attrib['rowType'] = meta_elem_attrib.meta_element_type.type.value
         elem.attrib['fieldsTerminatedBy'] = meta_elem_attrib.meta_element_type.csv_encoding.csv_delimiter
@@ -225,13 +225,13 @@ def _build_meta_xml(self, meta_elem_attrib: MetaElementAttributes):
         location = ET.SubElement(files, 'location')
         location.text = meta_elem_attrib.meta_element_type.file_name
         if meta_elem_attrib.core_id:
-            id_field = ET.SubElement(elem, MetaDefaultFields.ID) \
-                if meta_elem_attrib.meta_element_type.core_or_ext_type == 'core' \
-                else ET.SubElement(elem, MetaDefaultFields.CORE_ID)
+            id_field = ET.SubElement(elem, Defaults.MetaDefaultFields.ID) \
+                if meta_elem_attrib.meta_element_type.core_or_ext_type == CoreOrExtType.CORE \
+                else ET.SubElement(elem, Defaults.MetaDefaultFields.CORE_ID)
             id_field.attrib['index'] = meta_elem_attrib.core_id.index
 
         for _, f in enumerate(meta_elem_attrib.fields):
-            if f.field_name not in list(MetaDefaultFields):
+            if f.field_name not in list(Defaults.MetaDefaultFields):
                 field_elem = ET.SubElement(elem, "field")
                 if f.index is not None:
                     field_elem.attrib['index'] = f.index

src/dwcahandler/dwca/terms.py

@@ -16,7 +16,7 @@
 
 
 def absolute_file_paths(directory):
-    """Convert files in a directory into absolute paths and return
+    """Convert data in a directory into absolute paths and return
     as a generator
 
     :param directory: The directory to scan.
@@ -71,7 +71,7 @@ class Terms:
 
     GBIF_EXT = "https://rs.gbif.org/extensions.json"
 
-    GBIF_REGISTERED_EXTENSION = [e for e in GbifRegisteredExt]
+    GBIF_REGISTERED_EXTENSION = pd.DataFrame(columns=["prefix", "identifier", "namespace", "issued_date"])#[e for e in GbifRegisteredExt]
 
     DWC_SOURCE_URL = "https://raw.githubusercontent.com/tdwg/rs.tdwg.org/master/terms/terms.csv"
 
@@ -267,7 +267,7 @@ def __sort_values(df_to_sort: pd.DataFrame, sorting_column: str) -> pd.DataFrame
 
         log.info("Current class and terms")
 
-        exclude_update_prefixes = [NsPrefix.DC.value]
+        exclude_update_prefixes = [NsPrefix.DC.value, NsPrefix.DWC.value]
         terms = Terms()
         print(terms.class_df.groupby(["prefix"]).agg(
             class_prefix_count=pd.NamedAgg(column="prefix", aggfunc="count")
@@ -277,7 +277,7 @@ def __sort_values(df_to_sort: pd.DataFrame, sorting_column: str) -> pd.DataFrame
         ))
         terms.class_df = terms.class_df[terms.class_df.prefix.isin(exclude_update_prefixes)]
         terms.terms_df = terms.terms_df[terms.terms_df.prefix.isin(exclude_update_prefixes)]
-        terms.update_dwc_terms()
+        #terms.update_dwc_terms()
         terms.update_gbif_ext()
         terms.class_df = __sort_values(terms.class_df, "class")
         terms.terms_df = __sort_values(terms.terms_df, "term")
@@ -291,3 +291,5 @@ def __sort_values(df_to_sort: pd.DataFrame, sorting_column: str) -> pd.DataFrame
             term_prefix_count=pd.NamedAgg(column="prefix", aggfunc="count")
         ))
         return terms.terms_df, terms.class_df
+
+#Terms.update_terms()

tests/__init__.py

@@ -5,7 +5,7 @@
 import csv
 from dwcahandler import Eml
 from xml.dom.minidom import parseString
-from dwcahandler import MetaDwCA, MetaDefaultFields
+from dwcahandler import MetaDwCA, Defaults
 
 
 def get_eml_content():
@@ -25,7 +25,7 @@ def make_fields(columns: list, term_uri: str, field_start: int = 0, core_id: str
         idx_start = field_start if field_start != -2 else 0
 
     for idx, col in enumerate(columns):
-        if not (col in list(MetaDefaultFields)):
+        if not (col in list(Defaults.MetaDefaultFields)):
             dwc_term_uri = "http://rs.tdwg.org/dwc/terms" if col == 'occurrenceID' else term_uri
             fields = fields + '\n' + f'<field index="{str(idx + idx_start)}" term="{dwc_term_uri}/{col}"/>'
 

tests/test_write_dwca.py

@@ -51,7 +51,7 @@ def test_generate_dwca_without_ext(self):
                 root = tree.getroot()
                 ns = _get_namespace(root)
                 assert ns == "{http://rs.tdwg.org/dwc/text/}"
-                core_node = root.find(f'{ns}{CoreOrExtType.CORE}')
+                core_node = root.find(f'{ns}{CoreOrExtType.CORE.value}')
                 assert core_node is not None
                 fields = core_node.findall(f'{ns}field')
                 term_fields = [f.attrib.get('term') for f in fields]
@@ -90,7 +90,7 @@ def test_generate_dwca_with_ext(self):
                 root = tree.getroot()
                 ns = _get_namespace(root)
                 assert ns == "{http://rs.tdwg.org/dwc/text/}"
-                core_node = root.find(f'{ns}{CoreOrExtType.CORE}')
+                core_node = root.find(f'{ns}{CoreOrExtType.CORE.value}')
                 assert core_node is not None
                 fields = core_node.findall(f'{ns}field')
                 term_fields = [f.attrib.get('term') for f in fields]
@@ -99,7 +99,7 @@ def test_generate_dwca_with_ext(self):
                     assert any(sample_col in f for f in term_fields)
                 core_file = core_node.find(f'{ns}files').find(f'{ns}location').text
 
-                ext_node = root.find(f'{ns}{CoreOrExtType.EXTENSION}')
+                ext_node = root.find(f'{ns}{CoreOrExtType.EXTENSION.value}')
                 assert ext_node is not None
                 fields = ext_node.findall(f'{ns}field')
                 term_fields = [f.attrib.get('term') for f in fields]
@@ -150,7 +150,7 @@ def test_generate_dwca_in_memory(self):
                 root = tree.getroot()
                 ns = _get_namespace(root)
                 assert ns == "{http://rs.tdwg.org/dwc/text/}"
-                core_node = root.find(f'{ns}{CoreOrExtType.CORE}')
+                core_node = root.find(f'{ns}{CoreOrExtType.CORE.value}')
                 assert core_node is not None
                 fields = core_node.findall(f'{ns}field')
                 term_fields = [f.attrib.get('term') for f in fields]