Merge pull request #47 from getindata/release-0.21.0

Release 0.21.0

Author: swtwsk

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+## [0.21.0] - 2022-02-11
+
+This version brings compatibility with `dbt 1.0`.
+
 ## [0.20.1] - 2022-02-08
 
 ## [0.20.0] - 2022-02-08
@@ -94,7 +98,9 @@
 
 -   Initial implementation of `dbt_airflow_manifest_parser` library.
 
-[Unreleased]: https://github.com/getindata/dbt-airflow-factory/compare/0.20.1...HEAD
+[Unreleased]: https://github.com/getindata/dbt-airflow-factory/compare/0.21.0...HEAD
+
+[0.21.0]: https://github.com/getindata/dbt-airflow-factory/compare/0.20.1...0.21.0
 
 [0.20.1]: https://github.com/getindata/dbt-airflow-factory/compare/0.20.0...0.20.1
 

dbt_airflow_factory/__init__.py

@@ -1 +1 @@
-version = "0.20.1"
+version = "0.21.0"

dbt_airflow_factory/builder.py

@@ -64,6 +64,7 @@ def _make_dbt_test_task(self, model_name: str, is_in_task_group: bool) -> BaseOp
             self._build_task_name(model_name, command, is_in_task_group),
             command,
             model_name,
+            additional_dbt_args=["--indirect-selection=cautious"],
         )
 
     def _make_dbt_multiple_deps_test_task(

dbt_airflow_factory/ecs/ecs_operator.py

@@ -1,6 +1,6 @@
 """Factories creating Airflow Operators running DBT tasks."""
 
-from typing import Optional
+from typing import List, Optional
 
 from airflow.models.baseoperator import BaseOperator
 
@@ -24,5 +24,11 @@ def __init__(
         self.dbt_execution_env_parameters = dbt_execution_env_parameters
         self.ecs_execution_parameters = ecs_execution_parameters
 
-    def create(self, name: str, command: str, model: Optional[str] = None) -> BaseOperator:
+    def create(
+        self,
+        name: str,
+        command: str,
+        model: Optional[str] = None,
+        additional_dbt_args: Optional[List[str]] = None,
+    ) -> BaseOperator:
         raise NotImplementedError  # TODO

dbt_airflow_factory/k8s/k8s_operator.py

@@ -42,10 +42,18 @@ def __init__(
         self.dbt_execution_env_parameters = dbt_execution_env_parameters
         self.kubernetes_execution_parameters = kubernetes_execution_parameters
 
-    def create(self, name: str, command: str, model: Optional[str] = None) -> BaseOperator:
-        return self._create(self._prepare_arguments(command, model), name)
+    def create(
+        self,
+        name: str,
+        command: str,
+        model: Optional[str] = None,
+        additional_dbt_args: Optional[List[str]] = None,
+    ) -> BaseOperator:
+        return self._create(self._prepare_arguments(command, model, additional_dbt_args), name)
 
-    def _prepare_arguments(self, command: str, model: Optional[str]) -> List[str]:
+    def _prepare_arguments(
+        self, command: str, model: Optional[str], additional_dbt_args: Optional[List[str]]
+    ) -> List[str]:
         args = [
             "set -e;",
             f"dbt --no-write-json {command}",
@@ -55,7 +63,9 @@ def _prepare_arguments(self, command: str, model: Optional[str]) -> List[str]:
             f"--profiles-dir {self.dbt_execution_env_parameters.profile_dir_path}",
         ]
         if model:
-            args += [f"--models {model}"]
+            args += [f"--select {model}"]
+        if additional_dbt_args:
+            args += additional_dbt_args
         return [" ".join(args)]
 
     def _create(self, args: Optional[List[str]], name: str) -> KubernetesPodOperator:

dbt_airflow_factory/operator.py

@@ -1,7 +1,7 @@
 """Factories creating Airflow Operators running DBT tasks."""
 
 import abc
-from typing import Optional
+from typing import List, Optional
 
 import airflow
 
@@ -20,7 +20,13 @@ class DbtRunOperatorBuilder(metaclass=abc.ABCMeta):
     """
 
     @abc.abstractmethod
-    def create(self, name: str, command: str, model: Optional[str] = None) -> BaseOperator:
+    def create(
+        self,
+        name: str,
+        command: str,
+        model: Optional[str] = None,
+        additional_dbt_args: Optional[List[str]] = None,
+    ) -> BaseOperator:
         """
         Create Airflow Operator running a single DBT task.
 
@@ -30,6 +36,8 @@ def create(self, name: str, command: str, model: Optional[str] = None) -> BaseOp
         :type command: str
         :param model: models to include.
         :type model: Optional[str]
+        :param additional_dbt_args: Additional arguments to pass to dbt.
+        :type additional_dbt_args: Optional[List[str]]
         :return: Airflow Operator running a single DBT task.
         :rtype: BaseOperator
         """

setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.20.1
+current_version = 0.21.0
 
 [bumpversion:file:setup.py]
 

setup.py

@@ -29,7 +29,7 @@
 
 setup(
     name="dbt-airflow-factory",
-    version="0.20.1",
+    version="0.21.0",
     description="Library to convert DBT manifest metadata to Airflow tasks",
     long_description=README,
     long_description_content_type="text/markdown",

tests/manifest.json

@@ -1,7 +1,7 @@
 {
   "metadata": {
     "dbt_schema_version": "https://schemas.getdbt.com/dbt/manifest/v3.json",
-    "dbt_version": "0.21.0",
+    "dbt_version": "1.0.0",
     "generated_at": "2021-10-19T09:11:11.955603Z",
     "invocation_id": "a3958a44-f911-4921-a78c-5bb8d512ac25",
     "env": {},

tests/test_dependencies.py

@@ -228,7 +228,7 @@ def test_complex_test_dependencies():
     )
 
     def extract_model_arguments(args: str) -> List[str]:
-        return list(filter(lambda s: not s.startswith("-"), args.split("--models ")[1].split()))
+        return list(filter(lambda s: not s.startswith("-"), args.split("--select ")[1].split()))
 
     assert (
         "model2_model3_test"

tests/test_tasks.py

@@ -38,7 +38,7 @@ def test_run_task():
     run_task = tasks.get_task("model.dbt_test.dim_users").run_airflow_task
     assert run_task.cmds == ["bash", "-c"]
     assert "set -e; dbt --no-write-json run " in run_task.arguments[0]
-    assert "--models dim_users" in run_task.arguments[0]
+    assert "--select dim_users" in run_task.arguments[0]
     assert '--vars "{}"' in run_task.arguments[0]
     assert run_task.name == "dim-users-run" if IS_FIRST_AIRFLOW_VERSION else "run"
     assert run_task.task_id == task_group_prefix_builder("dim_users", "run")
@@ -56,7 +56,7 @@ def test_test_task():
     test_task = tasks.get_task("model.dbt_test.dim_users").test_airflow_task
     assert test_task.cmds == ["bash", "-c"]
     assert "set -e; dbt --no-write-json test " in test_task.arguments[0]
-    assert "--models dim_users" in test_task.arguments[0]
+    assert "--select dim_users" in test_task.arguments[0]
     assert '--vars "{}"' in test_task.arguments[0]
     assert test_task.name == "dim-users-test" if IS_FIRST_AIRFLOW_VERSION else "test"
     assert test_task.task_id == task_group_prefix_builder("dim_users", "test")