refactor: Move schema matchers to a higher directory level for better organization

Author: roquelopez

bdikit/api.py

@@ -7,10 +7,11 @@
 import panel as pn
 from IPython.display import display, Markdown
 
-from bdikit.schema_matching.one2one.base import BaseSchemaMatcher
-from bdikit.schema_matching.one2one.matcher_factory import SchemaMatchers
-from bdikit.schema_matching.topk.base import BaseTopkSchemaMatcher
-from bdikit.schema_matching.topk.matcher_factory import TopkMatchers
+from bdikit.schema_matching.base import BaseOne2oneSchemaMatcher, BaseTopkSchemaMatcher
+from bdikit.schema_matching.matcher_factory import (
+    get_one2one_schema_matcher,
+    get_topk_schema_matcher,
+)
 from bdikit.value_matching.base import BaseValueMatcher, ValueMatch, ValueMatchingResult
 from bdikit.value_matching.matcher_factory import ValueMatchers
 from bdikit.standards.standard_factory import Standards
@@ -43,7 +44,7 @@
 def match_schema(
     source: pd.DataFrame,
     target: Union[str, pd.DataFrame] = "gdc",
-    method: Union[str, BaseSchemaMatcher] = DEFAULT_SCHEMA_MATCHING_METHOD,
+    method: Union[str, BaseOne2oneSchemaMatcher] = DEFAULT_SCHEMA_MATCHING_METHOD,
     method_args: Optional[Dict[str, Any]] = None,
     standard_args: Optional[Dict[str, Any]] = None,
 ) -> pd.DataFrame:
@@ -74,15 +75,15 @@ def match_schema(
     if isinstance(method, str):
         if method_args is None:
             method_args = {}
-        matcher_instance = SchemaMatchers.get_matcher(method, **method_args)
-    elif isinstance(method, BaseSchemaMatcher):
+        matcher_instance = get_one2one_schema_matcher(method, **method_args)
+    elif isinstance(method, BaseOne2oneSchemaMatcher):
         matcher_instance = method
     else:
         raise ValueError(
             "The method must be a string or an instance of BaseColumnMappingAlgorithm"
         )
 
-    matches = matcher_instance.map(source, target_table)
+    matches = matcher_instance.get_one2one_match(source, target_table)
 
     return pd.DataFrame(matches.items(), columns=["source", "target"])
 
@@ -138,15 +139,15 @@ def top_matches(
     if isinstance(method, str):
         if method_args is None:
             method_args = {}
-        topk_matcher = TopkMatchers.get_matcher(method, **method_args)
+        topk_matcher = get_topk_schema_matcher(method, **method_args)
     elif isinstance(method, BaseTopkSchemaMatcher):
         topk_matcher = method
     else:
         raise ValueError(
             "The method must be a string or an instance of BaseTopkColumnMatcher"
         )
 
-    top_k_matches = topk_matcher.get_recommendations(
+    top_k_matches = topk_matcher.get_topk_matches(
         selected_columns, target=target_table, top_k=top_k
     )
 

bdikit/schema_matching/topk/base.py bdikit/schema_matching/base.py

@@ -1,8 +1,22 @@
-from bdikit.schema_matching.one2one.base import BaseSchemaMatcher
 from typing import List, NamedTuple, TypedDict, Dict
 import pandas as pd
 
 
+class BaseOne2oneSchemaMatcher:
+    def get_one2one_match(
+        self, source: pd.DataFrame, target: pd.DataFrame
+    ) -> Dict[str, str]:
+        raise NotImplementedError("Subclasses must implement this method")
+
+    def _fill_missing_matches(
+        self, dataset: pd.DataFrame, matches: Dict[str, str]
+    ) -> Dict[str, str]:
+        for column in dataset.columns:
+            if column not in matches:
+                matches[column] = ""
+        return matches
+
+
 class ColumnScore(NamedTuple):
     column_name: str
     score: float
@@ -13,19 +27,19 @@ class TopkMatching(TypedDict):
     top_k_columns: List[ColumnScore]
 
 
-class BaseTopkSchemaMatcher(BaseSchemaMatcher):
+class BaseTopkSchemaMatcher(BaseOne2oneSchemaMatcher):
 
-    def get_recommendations(
+    def get_topk_matches(
         self, source: pd.DataFrame, target: pd.DataFrame, top_k: int
     ) -> List[TopkMatching]:
         raise NotImplementedError("Subclasses must implement this method")
 
-    def map(
+    def get_one2one_match(
         self,
         source: pd.DataFrame,
         target: pd.DataFrame,
     ) -> Dict[str, str]:
-        top_matches = self.get_recommendations(source, target, 1)
+        top_matches = self.get_topk_matches(source, target, 1)
         matches = {}
 
         for top_match in top_matches:

bdikit/schema_matching/topk/contrastivelearning.py bdikit/schema_matching/contrastivelearning.py

@@ -1,7 +1,7 @@
 import pandas as pd
 import numpy as np
 from typing import List
-from bdikit.schema_matching.topk.base import (
+from bdikit.schema_matching.base import (
     ColumnScore,
     TopkMatching,
     BaseTopkSchemaMatcher,
@@ -14,12 +14,12 @@
 from bdikit.models import ColumnEmbedder
 
 
-class EmbeddingSimilarityTopkSchemaMatcher(BaseTopkSchemaMatcher):
+class EmbeddingSimilarity(BaseTopkSchemaMatcher):
     def __init__(self, column_embedder: ColumnEmbedder, metric: str = "cosine"):
         self.api = column_embedder
         self.metric = metric
 
-    def get_recommendations(
+    def get_topk_matches(
         self, source: pd.DataFrame, target: pd.DataFrame, top_k: int = 10
     ) -> List[TopkMatching]:
         """
@@ -54,7 +54,7 @@ def get_recommendations(
         return top_k_results
 
 
-class CLTopkSchemaMatcher(EmbeddingSimilarityTopkSchemaMatcher):
+class ContrastiveLearning(EmbeddingSimilarity):
     def __init__(self, model_name: str = DEFAULT_CL_MODEL, metric: str = "cosine"):
         super().__init__(
             column_embedder=ContrastiveLearningAPI(model_name=model_name), metric=metric

bdikit/schema_matching/one2one/gpt.py bdikit/schema_matching/gpt.py

@@ -1,13 +1,13 @@
 import pandas as pd
 from openai import OpenAI
-from bdikit.schema_matching.one2one.base import BaseSchemaMatcher
+from bdikit.schema_matching.base import BaseOne2oneSchemaMatcher
 
 
-class GPTSchemaMatcher(BaseSchemaMatcher):
+class GPT(BaseOne2oneSchemaMatcher):
     def __init__(self):
         self.client = OpenAI()
 
-    def map(self, source: pd.DataFrame, target: pd.DataFrame):
+    def get_one2one_match(self, source: pd.DataFrame, target: pd.DataFrame):
         target_columns = target.columns
         labels = ", ".join(target_columns)
         candidate_columns = source.columns

bdikit/schema_matching/topk/magneto.py bdikit/schema_matching/magneto.py

@@ -1,13 +1,8 @@
 import pandas as pd
 from typing import Dict, Any, List
 from magneto import Magneto as Magneto_Lib
-from bdikit.schema_matching.one2one.base import BaseSchemaMatcher
 from bdikit.download import get_cached_model_or_download
-from bdikit.schema_matching.topk.base import (
-    ColumnScore,
-    TopkMatching,
-    BaseTopkSchemaMatcher,
-)
+from bdikit.schema_matching.base import ColumnScore, TopkMatching, BaseTopkSchemaMatcher
 
 DEFAULT_MAGNETO_MODEL = "magneto-gdc-v0.1"
 
@@ -18,7 +13,7 @@ def __init__(self, kwargs: Dict[str, Any] = None):
             kwargs = {}
         self.magneto = Magneto_Lib(**kwargs)
 
-    def get_recommendations(
+    def get_topk_matches(
         self, source: pd.DataFrame, target: pd.DataFrame, top_k: int
     ) -> List[TopkMatching]:
         self.magneto.params["topk"] = (

bdikit/schema_matching/matcher_factory.py

@@ -0,0 +1,114 @@
+import importlib
+from enum import Enum
+from typing import Mapping, Dict, Any
+from bdikit.schema_matching.base import BaseOne2oneSchemaMatcher, BaseTopkSchemaMatcher
+
+
+class One2oneSchemaMatchers(Enum):
+    SIMFLOOD = (
+        "similarity_flooding",
+        "bdikit.schema_matching.valentine.SimFlood",
+    )
+    COMA = (
+        "coma",
+        "bdikit.schema_matching.valentine.Coma",
+    )
+    CUPID = (
+        "cupid",
+        "bdikit.schema_matching.valentine.Cupid",
+    )
+    DISTRIBUTION_BASED = (
+        "distribution_based",
+        "bdikit.schema_matching.valentine.DistributionBased",
+    )
+    JACCARD_DISTANCE = (
+        "jaccard_distance",
+        "bdikit.schema_matching.valentine.Jaccard",
+    )
+    GPT = ("gpt", "bdikit.schema_matching.gpt.GPT")
+
+    TWO_PHASE = (
+        "two_phase",
+        "bdikit.schema_matching.twophase.TwoPhase",
+    )
+
+    def __init__(self, matcher_name: str, matcher_path: str):
+        self.matcher_name = matcher_name
+        self.matcher_path = matcher_path
+
+
+class TopkSchemaMatchers(Enum):
+    CT_LEARNING = (
+        "ct_learning",
+        "bdikit.schema_matching.contrastivelearning.ContrastiveLearning",
+    )
+
+    MAX_VAL_SIM = (
+        "max_val_sim",
+        "bdikit.schema_matching.topk.maxvalsim.MaxValSim",
+    )
+
+    MAGNETO_ZS_BP = (
+        "magneto_zs_bp",
+        "bdikit.schema_matching.magneto.MagnetoZSBP",
+    )
+
+    MAGNETO_FT_BP = (
+        "magneto_ft_bp",
+        "bdikit.schema_matching.magneto.MagnetoFTBP",
+    )
+
+    MAGNETO_ZS_LLM = (
+        "magneto_zs_llm",
+        "bdikit.schema_matching.magneto.MagnetoZSLLM",
+    )
+
+    MAGNETO_FT_LLM = (
+        "magneto_ft_llm",
+        "bdikit.schema_matching.magneto.MagnetoFTLLM",
+    )
+
+    def __init__(self, matcher_name: str, matcher_path: str):
+        self.matcher_name = matcher_name
+        self.matcher_path = matcher_path
+
+
+one2one_schema_matchers = {
+    method.matcher_name: method.matcher_path for method in One2oneSchemaMatchers
+}
+topk_schema_matchers = {
+    method.matcher_name: method.matcher_path for method in TopkSchemaMatchers
+}
+one2one_schema_matchers.update(topk_schema_matchers)
+
+
+def create_matcher(
+    matcher_name: str,
+    available_matchers: Dict[str, str],
+    **matcher_kwargs: Mapping[str, Any],
+):
+    if matcher_name not in available_matchers:
+        names = ", ".join(list(available_matchers.keys()))
+        raise ValueError(
+            f"The {matcher_name} algorithm is not supported. "
+            f"Supported algorithms are: {names}"
+        )
+    # Load the class dynamically
+    module_path, class_name = available_matchers[matcher_name].rsplit(".", 1)
+    module = importlib.import_module(module_path)
+
+    return getattr(module, class_name)(**matcher_kwargs)
+
+
+def get_one2one_schema_matcher(
+    matcher_name: str, **matcher_kwargs: Mapping[str, Any]
+) -> BaseOne2oneSchemaMatcher:
+
+    return create_matcher(matcher_name, one2one_schema_matchers, **matcher_kwargs)
+
+
+def get_topk_schema_matcher(
+    matcher_name: str, **matcher_kwargs: Mapping[str, Any]
+) -> BaseTopkSchemaMatcher:
+
+    return create_matcher(matcher_name, topk_schema_matchers, **matcher_kwargs)

bdikit/schema_matching/topk/maxvalsim.py bdikit/schema_matching/maxvalsim.py

@@ -1,17 +1,17 @@
 import pandas as pd
 from typing import Optional, List
 from bdikit.models.contrastive_learning.cl_api import DEFAULT_CL_MODEL
-from bdikit.schema_matching.topk.base import (
+from bdikit.schema_matching.base import (
     BaseTopkSchemaMatcher,
     TopkMatching,
     ColumnScore,
 )
-from bdikit.schema_matching.topk.contrastivelearning import CLTopkSchemaMatcher
+from bdikit.schema_matching.contrastivelearning import ContrastiveLearning
 from bdikit.value_matching.polyfuzz import TFIDFValueMatcher
 from bdikit.value_matching.base import BaseValueMatcher
 
 
-class MaxValSimSchemaMatcher(BaseTopkSchemaMatcher):
+class MaxValSim(BaseTopkSchemaMatcher):
     def __init__(
         self,
         top_k: int = 20,
@@ -20,7 +20,7 @@ def __init__(
         value_matcher: Optional[BaseValueMatcher] = None,
     ):
         if top_k_matcher is None:
-            self.api = CLTopkSchemaMatcher(DEFAULT_CL_MODEL)
+            self.api = ContrastiveLearning(DEFAULT_CL_MODEL)
         elif isinstance(top_k_matcher, BaseTopkSchemaMatcher):
             self.api = top_k_matcher
         else:
@@ -49,13 +49,13 @@ def unique_string_values(self, column: pd.Series) -> pd.Series:
         else:
             return pd.Series(column.unique().astype(str), name=column.name)
 
-    def get_recommendations(
+    def get_topk_matches(
         self, source: pd.DataFrame, target: pd.DataFrame, top_k: int
     ) -> List[TopkMatching]:
         max_topk = max(
             top_k, self.top_k
         )  # If self.top_k (method param) is smaller than the requested top_k, use top_k
-        topk_column_matches = self.api.get_recommendations(source, target, max_topk)
+        topk_column_matches = self.api.get_topk_matches(source, target, max_topk)
         matches = {}
         top_k_results = []