Merge pull request #109 from preset-io/fix_del

fix: prevent node from being excluded twice

src/preset_cli/cli/superset/sync/dbt/lib.py

@@ -386,20 +386,21 @@ def apply_select(
     """
     Apply dbt node selection (https://docs.getdbt.com/reference/node-selection/syntax).
     """
-    if not select:
-        return models
-
     model_ids = {model["unique_id"]: model for model in models}
 
-    selected: Dict[str, ModelSchema] = {}
-    for selection in select:
-        ids = set.intersection(
-            *[
-                {model["unique_id"] for model in filter_models(models, condition)}
-                for condition in selection.split(",")
-            ]
-        )
-        selected.update({id_: model_ids[id_] for id_ in ids})
+    selected: Dict[str, ModelSchema]
+    if not select:
+        selected = {model["unique_id"]: model for model in models}
+    else:
+        selected = {}
+        for selection in select:
+            ids = set.intersection(
+                *[
+                    {model["unique_id"] for model in filter_models(models, condition)}
+                    for condition in selection.split(",")
+                ]
+            )
+            selected.update({id_: model_ids[id_] for id_ in ids})
 
     for selection in exclude:
         for id_ in set.intersection(
@@ -408,6 +409,7 @@ def apply_select(
                 for condition in selection.split(",")
             ]
         ):
-            del selected[id_]
+            if id_ in selected:
+                del selected[id_]
 
     return list(selected.values())

tests/cli/superset/sync/dbt/lib_test.py

@@ -180,7 +180,7 @@ def test_build_snowflake_sqlalchemy_params_pk(fs: FakeFilesystem) -> None:
                     "privatekey_body": "-----BEGIN ENCRYPTED PRIVATE KEY",
                     "privatekey_pass": "XXX",
                 },
-            }
+            },
         ),
     }
 
@@ -437,10 +437,27 @@ def test_apply_select() -> None:
         "one",
     }
 
-    # test exclude
     assert {
         model["name"]
         for model in apply_select(models, ("+two+",), ("three", "tag:test"))
     } == {
         "two",
     }
+
+
+def test_apply_select_exclude() -> None:
+    """
+    Custom tests for the ``exclude`` option.
+    """
+    a = dict(name="a", tags=[], unique_id="a", depends_on=[], children=["b", "c"])
+    b = dict(name="b", tags=[], unique_id="b", depends_on=["a"], children=["d"])
+    c = dict(name="c", tags=[], unique_id="c", depends_on=["a"], children=["d"])
+    d = dict(name="d", tags=[], unique_id="d", depends_on=["b", "c"], children=[])
+    models: List[ModelSchema] = [a, b, c, d]  # type: ignore
+
+    assert {model["name"] for model in apply_select(models, (), ("d",))} == {
+        "a",
+        "b",
+        "c",
+    }
+    assert {model["name"] for model in apply_select(models, (), ("b+", "c+"))} == {"a"}