add more robust support for multi connections

metrics_layer/cli/seeding.py

@@ -266,8 +266,13 @@ def seed(self, auto_tag_searchable_fields: bool = False):
             dumper.dump_yaml_file(project_data, zenlytic_project_path)
 
     def get_model_name(self, current_models: list):
-        if len(current_models) > 0:
+        if len(current_models) == 1:
             return current_models[0].name
+        elif len(current_models) > 1:
+            for model in current_models:
+                if self.connection and model.connection == self.connection.name:
+                    return model.name
+            raise ValueError("Multiple models found, but none match the connection name")
         return self.default_model_name
 
     def make_models(self):

metrics_layer/core/model/field.py

@@ -2778,8 +2778,14 @@ def join_graphs(self):
             return base
 
         edges = self.view.project.join_graph.merged_results_graph(self.view.model).in_edges(self.id())
-        extended = [f"merged_result_{mr}" for mr, _ in edges]
+        extended = self._wrap_with_model_name([f"merged_result_{mr}" for mr, _ in edges])
 
         if self.loses_join_ability_with_other_views():
             return extended
         return list(sorted(base + extended))
+
+    def _wrap_with_model_name(self, join_graphs: list):
+        if len(models := self.view.project.models()) > 1:
+            model_index = [m.name for m in models].index(self.view.model.name)
+            return [f"m{model_index}_{jg}" for jg in join_graphs]
+        return join_graphs

metrics_layer/core/model/project.py

@@ -116,9 +116,13 @@ def remove_field(self, field_name: str, view_name: str, refresh_cache: bool = Tr
     def timezone(self):
         if self._timezone:
             return self._timezone
-        for m in self.models():
-            if m.timezone:
-                return m.timezone
+        timezones = list(set(m.timezone for m in self.models() if m.timezone))
+        if len(timezones) == 1:
+            return timezones[0]
+        elif len(timezones) > 1:
+            raise QueryError(
+                "Multiple timezones found in models, please specify only one timezone across models"
+            )
         return None
 
     @property

metrics_layer/core/sql/arbitrary_merge_resolve.py

@@ -37,6 +37,7 @@ def __init__(
         self.project = project
         self.connections = connections
         self.connection = None
+        self.model = None
         # All queries are merged queries (obviously)
         self.query_kind = QueryKindTypes.merged
         self.kwargs = kwargs
@@ -76,6 +77,7 @@ def get_query(self, semicolon: bool = True):
             )
 
         mapping_lookup = self._mapping_lookup
+        self.model = resolver.model
         clean_where = [{**w, "field": mapping_lookup.get(w["field"].lower(), w["field"])} for w in self.where]
         clean_having = [
             {**h, "field": mapping_lookup.get(h["field"].lower(), h["field"])} for h in self.having

metrics_layer/core/sql/resolve.py

@@ -1,4 +1,4 @@
-from collections import defaultdict
+from collections import Counter, defaultdict
 from copy import deepcopy
 from typing import List, Union
 
@@ -424,7 +424,7 @@ def _get_model_for_query(self, model_name: str = None, metrics: list = [], dimen
             return self._derive_model(metrics, dimensions)
 
     def _derive_model(self, metrics: list, dimensions: list):
-        all_model_names = []
+        all_model_names, mapping_model_names = [], []
         models = self.project.models()
         for f in metrics + dimensions:
             try:
@@ -434,18 +434,21 @@ def _derive_model(self, metrics: list, dimensions: list):
                 for model in models:
                     try:
                         self.project.get_mapped_field(f, model=model)
-                        all_model_names.append(model.name)
-                        break
+                        mapping_model_names.append(model.name)
                     except Exception:
                         pass
 
         all_model_names = list(set(all_model_names))
-
-        if len(all_model_names) == 0:
+        if len(all_model_names) == 0 and len(mapping_model_names) > 0:
             # In a case that there are no models in the query, we'll just use the first model
             # in the project. This case should be limited to only mapping-only queries, so this is safe.
-            return self.project.models()[0]
-        elif len(all_model_names) == 1:
+            model_counts = Counter(mapping_model_names)
+            sorted_models = [m for m, _ in model_counts.most_common()]
+            return self.project.get_model(sorted_models[0])
+        elif len(all_model_names) == 1 and (
+            len(mapping_model_names) == 0
+            or (len(mapping_model_names) > 0 and all_model_names[0] in mapping_model_names)
+        ):
             return self.project.get_model(list(all_model_names)[0])
         else:
             raise QueryError(

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "metrics_layer"
-version = "0.12.39"
+version = "0.12.40"
 description = "The open source metrics layer."
 authors = ["Paul Blankley <[email protected]>"]
 keywords = ["Metrics Layer", "Business Intelligence", "Analytics"]

tests/test_merged_results.py

@@ -150,20 +150,20 @@ def _blow_out_by_time_frame(join_graph: str, tf: list):
         "year",
     ]
     core_tf = ["raw", "time", "date", "week", "month", "quarter", "year"]
-    sub_q_cr = _blow_out_by_time_frame("merged_result_canon_date_core", core_tf)
-    sub_q_0_4 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_4", core_tf)
-    sub_q_0_2 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_7", core_tf)
-    sub_q_0_3 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_3", core_tf)
-    sub_q_0_5 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_5", core_tf)
-    sub_q_0_8 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_8", core_tf)
-    sub_q_0_9 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_9", core_tf)
-    sub_q_0_10 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_10", core_tf)
-    sub_q_0_11 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_11", core_tf)
-    sub_q_0_12 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_12", core_tf)
-    sub_q_0_14 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_13", core_tf)
-    sub_q_0_15 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_15", core_tf)
-    sub_q_0_16 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_16", core_tf)
-    sub_q_0_1 = _blow_out_by_time_frame("merged_result_subquery_0_subquery_1", core_tf)
+    sub_q_cr = _blow_out_by_time_frame("m0_merged_result_canon_date_core", core_tf)
+    sub_q_0_4 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_4", core_tf)
+    sub_q_0_2 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_7", core_tf)
+    sub_q_0_3 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_3", core_tf)
+    sub_q_0_5 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_5", core_tf)
+    sub_q_0_8 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_8", core_tf)
+    sub_q_0_9 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_9", core_tf)
+    sub_q_0_10 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_10", core_tf)
+    sub_q_0_11 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_11", core_tf)
+    sub_q_0_12 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_12", core_tf)
+    sub_q_0_14 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_13", core_tf)
+    sub_q_0_15 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_15", core_tf)
+    sub_q_0_16 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_16", core_tf)
+    sub_q_0_1 = _blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_1", core_tf)
     revenue_set = [
         *sub_q_cr,
         *sub_q_0_4,
@@ -189,47 +189,47 @@ def _blow_out_by_time_frame(join_graph: str, tf: list):
     field = connection.get_field("order_lines.order_date")
     order_lines_date_graphs = [
         "subquery_0",
-        "merged_result_canon_date_core_date",
-        "merged_result_subquery_0_subquery_10_date",
-        "merged_result_subquery_0_subquery_11_date",
-        "merged_result_subquery_0_subquery_12_date",
-        "merged_result_subquery_0_subquery_13_date",
-        "merged_result_subquery_0_subquery_15_date",
-        "merged_result_subquery_0_subquery_16_date",
-        "merged_result_subquery_0_subquery_1_date",
-        "merged_result_subquery_0_subquery_4_date",
-        "merged_result_subquery_0_subquery_7_date",
-        "merged_result_subquery_0_subquery_3_date",
-        "merged_result_subquery_0_subquery_5_date",
-        "merged_result_subquery_0_subquery_8_date",
-        "merged_result_subquery_0_subquery_9_date",
+        "m0_merged_result_canon_date_core_date",
+        "m0_merged_result_subquery_0_subquery_10_date",
+        "m0_merged_result_subquery_0_subquery_11_date",
+        "m0_merged_result_subquery_0_subquery_12_date",
+        "m0_merged_result_subquery_0_subquery_13_date",
+        "m0_merged_result_subquery_0_subquery_15_date",
+        "m0_merged_result_subquery_0_subquery_16_date",
+        "m0_merged_result_subquery_0_subquery_1_date",
+        "m0_merged_result_subquery_0_subquery_4_date",
+        "m0_merged_result_subquery_0_subquery_7_date",
+        "m0_merged_result_subquery_0_subquery_3_date",
+        "m0_merged_result_subquery_0_subquery_5_date",
+        "m0_merged_result_subquery_0_subquery_8_date",
+        "m0_merged_result_subquery_0_subquery_9_date",
     ]
     assert field.join_graphs() == list(sorted(order_lines_date_graphs))
 
     field = connection.get_field("orders.order_date")
     order_date_graphs = [
         "subquery_0",
-        "merged_result_canon_date_core_date",
-        "merged_result_subquery_0_subquery_10_date",
-        "merged_result_subquery_0_subquery_11_date",
-        "merged_result_subquery_0_subquery_12_date",
-        "merged_result_subquery_0_subquery_13_date",
-        "merged_result_subquery_0_subquery_15_date",
-        "merged_result_subquery_0_subquery_16_date",
-        "merged_result_subquery_0_subquery_1_date",
-        "merged_result_subquery_0_subquery_4_date",
-        "merged_result_subquery_0_subquery_7_date",
-        "merged_result_subquery_0_subquery_3_date",
-        "merged_result_subquery_0_subquery_5_date",
-        "merged_result_subquery_0_subquery_8_date",
-        "merged_result_subquery_0_subquery_9_date",
+        "m0_merged_result_canon_date_core_date",
+        "m0_merged_result_subquery_0_subquery_10_date",
+        "m0_merged_result_subquery_0_subquery_11_date",
+        "m0_merged_result_subquery_0_subquery_12_date",
+        "m0_merged_result_subquery_0_subquery_13_date",
+        "m0_merged_result_subquery_0_subquery_15_date",
+        "m0_merged_result_subquery_0_subquery_16_date",
+        "m0_merged_result_subquery_0_subquery_1_date",
+        "m0_merged_result_subquery_0_subquery_4_date",
+        "m0_merged_result_subquery_0_subquery_7_date",
+        "m0_merged_result_subquery_0_subquery_3_date",
+        "m0_merged_result_subquery_0_subquery_5_date",
+        "m0_merged_result_subquery_0_subquery_8_date",
+        "m0_merged_result_subquery_0_subquery_9_date",
     ]
     assert field.join_graphs() == list(sorted(order_date_graphs))
     field = connection.get_field("sub_channel")
 
     sub_channel_graphs = [
         "subquery_0",
-        *_blow_out_by_time_frame("merged_result_subquery_0_subquery_4", tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_4", tf),
     ]
     assert field.join_graphs() == list(sorted(sub_channel_graphs))
 
@@ -246,47 +246,47 @@ def _blow_out_by_time_frame(join_graph: str, tf: list):
         "subquery_11",
         "subquery_12",
         "subquery_13",
-        *_blow_out_by_time_frame("merged_result_subquery_0_subquery_1", tf),
-        *_blow_out_by_time_frame("merged_result_subquery_1_subquery_3", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_1_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_1_subquery_12", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_1_subquery_11", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_0_subquery_3", tf),
-        *_blow_out_by_time_frame("merged_result_subquery_0_subquery_12", tf),
-        *_blow_out_by_time_frame("merged_result_subquery_12_subquery_3", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_12_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_11_subquery_12", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_11_subquery_3", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_3_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_0_subquery_4", tf),
-        *_blow_out_by_time_frame("merged_result_subquery_0_subquery_11", tf),
-        *_blow_out_by_time_frame("merged_result_subquery_13_subquery_3", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_13_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_11_subquery_13", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_12_subquery_13", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_1_subquery_13", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_0_subquery_13", tf),
-        *_blow_out_by_time_frame("merged_result_subquery_11_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_1", tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_1_subquery_3", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_1_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_1_subquery_12", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_1_subquery_11", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_3", tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_12", tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_12_subquery_3", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_12_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_11_subquery_12", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_11_subquery_3", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_3_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_4", tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_11", tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_13_subquery_3", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_13_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_11_subquery_13", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_12_subquery_13", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_1_subquery_13", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_13", tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_11_subquery_4", core_tf),
     ]
     assert field.join_graphs() == list(sorted(gender_graphs))
 
     field = connection.get_field("number_of_sessions")
     sessions_graphs = [
         "subquery_4",
-        *_blow_out_by_time_frame("merged_result_canon_date_core", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_1_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_10_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_11_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_12_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_0_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_3_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_4_subquery_7", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_4_subquery_5", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_13_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_15_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_16_subquery_4", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_4_subquery_8", core_tf),
-        *_blow_out_by_time_frame("merged_result_subquery_4_subquery_9", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_canon_date_core", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_1_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_10_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_11_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_12_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_0_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_3_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_4_subquery_7", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_4_subquery_5", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_13_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_15_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_16_subquery_4", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_4_subquery_8", core_tf),
+        *_blow_out_by_time_frame("m0_merged_result_subquery_4_subquery_9", core_tf),
     ]
     assert field.join_graphs() == list(sorted(sessions_graphs))