From 7f4c1fe7b13f8bea65906e27e5712ff034daf01e Mon Sep 17 00:00:00 2001
From: Mike Gouline <1960272+gouline@users.noreply.github.com>
Date: Mon, 22 Aug 2022 16:10:37 +1000
Subject: [PATCH] Enable explicit foreign key relationships (#139)

---
 dbtmetabase/parsers/dbt_folder.py | 67 ++++++++++++++-----------------
 1 file changed, 31 insertions(+), 36 deletions(-)

diff --git a/dbtmetabase/parsers/dbt_folder.py b/dbtmetabase/parsers/dbt_folder.py
index a58aceae..1e9d3b13 100644
--- a/dbtmetabase/parsers/dbt_folder.py
+++ b/dbtmetabase/parsers/dbt_folder.py
@@ -1,7 +1,7 @@
 import re
 import yaml
 from pathlib import Path
-from typing import List, Iterable, Mapping, MutableMapping, Optional, Tuple
+from typing import List, Mapping, MutableMapping, Optional, Tuple
 
 from ..models.metabase import MetabaseModel, MetabaseColumn, ModelType
 from ..logger.logging import logger
@@ -181,50 +181,45 @@ def _read_column(self, column: Mapping, schema: str) -> MetabaseColumn:
         meta = column.get("meta", {})
         display_name = meta.get("metabase.display_name")
 
+        # Set explicitly (relationships override this)
+        fk_to = meta.get("metabase.foreign_key_to")
+        fk_field = meta.get("metabase.foreign_key_field")
+
         metabase_column = MetabaseColumn(
             name=column_name,
             description=column_description,
             display_name=display_name,
         )
 
-        tests: Optional[Iterable] = column.get("tests", [])
-        if tests is None:
-            tests = []
-
-        for test in tests:
+        for test in column.get("tests") or []:
             if isinstance(test, dict):
                 if "relationships" in test:
                     relationships = test["relationships"]
-                    parsed_table_ref = self.parse_ref(relationships["to"])
-                    if not parsed_table_ref:
-                        logger().warning(
-                            "Could not resolve foreign key target table for column %s",
-                            metabase_column.name,
-                        )
-                        continue
-
-                    parsed_ref = ".".join(
-                        map(
-                            lambda s: s.strip('"'),
-                            column.get("meta", {})
-                            .get("metabase.foreign_key_target_table", "")
-                            .split("."),
-                        )
-                    )
-                    if not parsed_ref or "." not in parsed_ref:
-                        parsed_ref = f"{schema}.{parsed_table_ref}"
-
-                    metabase_column.semantic_type = "type/FK"
-                    metabase_column.fk_target_table = parsed_ref.upper()
-                    metabase_column.fk_target_field = (
-                        str(relationships["field"]).upper().strip('"')
-                    )
-                    logger().debug(
-                        "Relation from %s to %s.%s",
-                        column.get("name", "").upper().strip('"'),
-                        metabase_column.fk_target_table,
-                        metabase_column.fk_target_field,
-                    )
+                    fk_to = relationships["to"]
+                    fk_field = relationships["field"]
+
+        if fk_to and fk_field:
+            fk_table = self.parse_ref(fk_to)
+            if fk_table:
+                metabase_column.semantic_type = "type/FK"
+                metabase_column.fk_target_table = f"{schema}.{fk_table}".upper()
+                metabase_column.fk_target_field = str(fk_field).upper().strip('"')
+                logger().debug(
+                    "Relation from %s to %s.%s",
+                    column.get("name", "").upper().strip('"'),
+                    metabase_column.fk_target_table,
+                    metabase_column.fk_target_field,
+                )
+            else:
+                logger().warning(
+                    "Could not resolve foreign key target table for column %s",
+                    metabase_column.name,
+                )
+        elif fk_to or fk_field:
+            logger().warning(
+                "Foreign key 'to' and 'field' must be provided for column %s",
+                metabase_column.name,
+            )
 
         for field, value in DbtReader.read_meta_fields(column).items():
             setattr(metabase_column, field, value)