refactor: use LabelMatching for label<>field matching

Author: pierrecamilleri

frictionless/table/header.py

@@ -7,6 +7,7 @@
 from .. import errors, helpers, types
 from ..exception import FrictionlessException
 from ..schema import Field
+from .label_matching import LabelMatching
 
 # The `fieldsMatch` modes are told apart by which mismatch they tolerate: a
 # label with no matching field, or a declared field with no matching label
@@ -25,7 +26,7 @@ class Header(List[str]):  # type: ignore
     > Constructor of this object is not Public API
 
     > Deprecated: using a `Header` directly as a list is deprecated, as the list
-    > ambiguously holds the names of the *schema* fieldss. Use its properties instead:
+    > ambiguously holds the names of the *schema* fields. Use its properties instead:
     > `labels` for the header row as read from the data source, `fields`/`field_names`
     > for the schema fields.
 
@@ -64,6 +65,7 @@ def __init__(
         self.__labels = labels
         self.__errors: List[errors.HeaderError] = []
         self.__expected_fields: Optional[List[Field]] = None
+        self.__matching = LabelMatching(labels, self.__fields, ignore_case=ignore_case)
         self.__process()
 
     # Deprecated
@@ -201,7 +203,7 @@ def get_expected_fields(self) -> List[Field]:
 
         expected: List[Field] = []
         for label in self.__labels:
-            field = self.__find_field_by_name(label)
+            field = self.__matching.matching_field(label)
             if field is None:
                 field = Field.from_descriptor({"name": label, "type": "any"})
             expected.append(field)
@@ -233,7 +235,7 @@ def _get_extra_labels(self) -> List[Tuple[int, str]]:
         return [
             (number, label)
             for number, label in enumerate(labels, start=1)
-            if self.__find_field_by_name(label) is None
+            if self.__matching.matching_field(label) is None
         ]
 
     def _get_missing_fields(self) -> List[Tuple[int, Field]]:
@@ -257,39 +259,19 @@ def _get_missing_fields(self) -> List[Tuple[int, Field]]:
         if not self.__matches_by_name:
             missing = fields[len(labels) :] if len(fields) > len(labels) else []
         else:
-            normalized_labels = [self.__normalize(label) for label in labels]
-
-            def is_absent(field: Field) -> bool:
-                return self.__normalize(field.name) not in normalized_labels
 
             def is_required(field: Field) -> bool:
                 return field.required or (
                     field.schema is not None and field.name in field.schema.primary_key
                 )
 
-            missing = [field for field in fields if is_absent(field)]
+            missing = self.__matching.unmatched_fields
             if self.__fields_match in TOLERATES_MISSING_FIELDS:
                 missing = [field for field in missing if is_required(field)]
 
         start = len(labels) + 1
         return [(start + offset, field) for offset, field in enumerate(missing)]
 
-    def __has_matching_field(self) -> bool:
-        """Whether at least one label corresponds to a schema field"""
-        return any(
-            self.__find_field_by_name(label) is not None for label in self.__labels
-        )
-
-    def __find_field_by_name(self, name: str) -> Optional[Field]:
-        target = self.__normalize(name)
-        for f in self.__fields:
-            if self.__normalize(f.name) == target:
-                return f
-        return None
-
-    def __normalize(self, s: str) -> str:
-        return s.lower() if self.__ignore_case else s
-
     # Convert
 
     def to_str(self):
@@ -330,11 +312,7 @@ def __process(self):
             )
 
         # Unmatched header
-        if (
-            self.__fields_match == "partial"
-            and fields
-            and not self.__has_matching_field()
-        ):
+        if self.__fields_match == "partial" and fields and not self.__matching.has_match:
             self.__errors.append(
                 errors.UnmatchedHeaderError(
                     note="",

frictionless/table/label_matching.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from typing import Dict, List, Optional
+
+from ..schema import Field
+
+
+class LabelMatching:
+    """Pairs the labels read from the data source with the schema fields, by name.
+
+    Parameters:
+        labels (str[]): the header row as read from the data source
+        fields (Field[]): the fields declared in the schema, in schema order
+        ignore_case (bool): compare labels and field names case-insensitively
+    """
+
+    def __init__(
+        self,
+        labels: List[str],
+        fields: List[Field],
+        *,
+        ignore_case: bool = False,
+    ) -> None:
+        self.__labels = labels
+        self.__fields = fields
+        self.__ignore_case = ignore_case
+
+        # Keyed by normalized name, in schema order; the first field wins in
+        # case of duplicates under normalization, so the duplicate fields are
+        # lost
+        fields_by_key: Dict[str, Field] = {}
+        for field in fields:
+            fields_by_key.setdefault(self.__normalize(field.name), field)
+
+        self.__fields_by_key = fields_by_key
+
+    def matching_field(self, label: str) -> Optional[Field]:
+        """Returns the field the given label matches, or None if there is none"""
+        return self.__fields_by_key.get(self.__normalize(label))
+
+    @property
+    def unmatched_fields(self) -> List[Field]:
+        """The fields no label matches, in schema order"""
+        matched = {self.__normalize(label) for label in self.__labels}
+        return [
+            field
+            for field in self.__fields
+            if self.__normalize(field.name) not in matched
+        ]
+
+    @property
+    def has_match(self) -> bool:
+        """Whether at least one label matches a schema field"""
+        return any(self.matching_field(label) is not None for label in self.__labels)
+
+    def __normalize(self, name: str) -> str:
+        """The normalized value a label and a field name are compared through"""
+        return name.lower() if self.__ignore_case else name