diff --git a/README.md b/README.md index b1b0f567..5a9b89b8 100644 --- a/README.md +++ b/README.md @@ -64,7 +64,7 @@ uv pip install . Add *airflow-dbt-python* to your `requirements.txt` file and edit your Airflow environment to use this new `requirements.txt` file, or upload it as a plugin. -Read the [documentation](https://airflow-dbt-python.readthedocs.io/en/latest/getting_started.html#installing-in-mwaa) for more a more detailed AWS MWAA installation breakdown. +Read the [documentation](https://airflow-dbt-python.readthedocs.io/en/latest/getting_started/#installing-in-mwaa) for more a more detailed AWS MWAA installation breakdown. ## In GCP Cloud Composer @@ -198,7 +198,7 @@ with DAG( ) as dag: dbt_test = DbtTestOperator( task_id="dbt_test", - selector_name="pre-run-tests", + selector="pre-run-tests", ) dbt_seed = DbtSeedOperator( @@ -221,7 +221,7 @@ More examples can be found in the [`examples/`](examples/) directory and the [do # Development -See the [development documentation](https://airflow-dbt-python.readthedocs.io/en/latest/development.html) for a more in-depth dive into setting up a development environment, running the test-suite, and general commentary on working on *airflow-dbt-python*. +See the [development documentation](https://airflow-dbt-python.readthedocs.io/en/latest/development/) for a more in-depth dive into setting up a development environment, running the test-suite, and general commentary on working on *airflow-dbt-python*. ## Testing diff --git a/airflow_dbt_python/operators/dbt.py b/airflow_dbt_python/operators/dbt.py index 5f49912e..6d8f8e2a 100644 --- a/airflow_dbt_python/operators/dbt.py +++ b/airflow_dbt_python/operators/dbt.py @@ -400,14 +400,15 @@ def __init__( full_refresh: Optional[bool] = None, models: Optional[list[str]] = None, select: Optional[list[str]] = None, - selector_name: Optional[list[str]] = None, + selector: Optional[str] = None, + selector_name: Optional[str] = None, exclude: Optional[list[str]] = None, **kwargs, ) -> None: super().__init__(**kwargs) self.full_refresh = full_refresh self.exclude = exclude - self.selector_name = selector_name + self.selector = selector or selector_name self.select = select or models @property @@ -432,6 +433,7 @@ def __init__( select: Optional[list[str]] = None, show: Optional[bool] = None, exclude: Optional[list[str]] = None, + selector: Optional[str] = None, selector_name: Optional[str] = None, **kwargs, ) -> None: @@ -440,7 +442,7 @@ def __init__( self.select = select self.show = show self.exclude = exclude - self.selector_name = selector_name + self.selector = selector or selector_name @property def command(self) -> str: @@ -465,6 +467,7 @@ def __init__( models: Optional[list[str]] = None, select: Optional[list[str]] = None, exclude: Optional[list[str]] = None, + selector: Optional[str] = None, selector_name: Optional[str] = None, indirect_selection: Optional[str] = None, **kwargs, @@ -473,7 +476,7 @@ def __init__( self.singular = singular self.generic = generic self.exclude = exclude - self.selector_name = selector_name + self.selector = selector or selector_name self.select = select or models self.indirect_selection = indirect_selection @@ -500,6 +503,7 @@ def __init__( models: Optional[list[str]] = None, select: Optional[list[str]] = None, exclude: Optional[list[str]] = None, + selector: Optional[str] = None, selector_name: Optional[str] = None, upload_dbt_project: bool = True, **kwargs, @@ -508,7 +512,7 @@ def __init__( self.parse_only = parse_only self.full_refresh = full_refresh self.exclude = exclude - self.selector_name = selector_name + self.selector = selector or selector_name self.select = select or models self.upload_dbt_project = upload_dbt_project @@ -614,13 +618,14 @@ def __init__( self, select: Optional[list[str]] = None, exclude: Optional[list[str]] = None, + selector: Optional[str] = None, selector_name: Optional[str] = None, **kwargs, ) -> None: super().__init__(**kwargs) self.select = select self.exclude = exclude - self.selector_name = selector_name + self.selector = selector or selector_name @property def command(self) -> str: @@ -644,6 +649,7 @@ def __init__( resource_types: Optional[list[str]] = None, select: Optional[list[str]] = None, exclude: Optional[list[str]] = None, + selector: Optional[str] = None, selector_name: Optional[str] = None, dbt_output: Optional[Output] = None, output_keys: Optional[list[str]] = None, @@ -654,7 +660,7 @@ def __init__( self.resource_types = resource_types self.select = select self.exclude = exclude - self.selector_name = selector_name + self.selector = selector or selector_name self.dbt_output = dbt_output self.output_keys = output_keys self.indirect_selection = indirect_selection @@ -727,6 +733,7 @@ def __init__( select: Optional[list[str]] = None, dbt_output: Optional[Union[str, Path]] = None, exclude: Optional[list[str]] = None, + selector: Optional[str] = None, selector_name: Optional[str] = None, upload_dbt_project: bool = True, **kwargs, @@ -734,7 +741,7 @@ def __init__( super().__init__(**kwargs) self.select = select self.exclude = exclude - self.selector_name = selector_name + self.selector = selector or selector_name self.dbt_output = dbt_output self.upload_dbt_project = upload_dbt_project @@ -759,6 +766,7 @@ def __init__( full_refresh: Optional[bool] = None, select: Optional[list[str]] = None, exclude: Optional[list[str]] = None, + selector: Optional[str] = None, selector_name: Optional[str] = None, singular: Optional[bool] = None, generic: Optional[bool] = None, @@ -770,7 +778,7 @@ def __init__( self.full_refresh = full_refresh self.select = select self.exclude = exclude - self.selector_name = selector_name + self.selector = selector or selector_name self.singular = singular self.generic = generic self.show = show diff --git a/airflow_dbt_python/utils/configs.py b/airflow_dbt_python/utils/configs.py index cda2bb5a..25285dd4 100644 --- a/airflow_dbt_python/utils/configs.py +++ b/airflow_dbt_python/utils/configs.py @@ -518,11 +518,13 @@ class SelectionConfig(BaseConfig): exclude: Optional[list[str]] = None select: Optional[list[str]] = None - selector_name: Optional[list[str]] = None selector: Optional[str] = None state: Optional[Union[Path, str]] = None defer_state: Optional[Union[Path, str]] = None + # Kept for compatibility with dbt versions < 1.5 + selector_name: Optional[str] = None + # Kept for compatibility with dbt versions < 0.21 models: Optional[list[str]] = None diff --git a/docs/development.rst b/docs/development.rst index 9489f946..ced6e861 100644 --- a/docs/development.rst +++ b/docs/development.rst @@ -26,11 +26,17 @@ Create a local virtual environment (if you don't want *uv* to manage it), and en .. code-block:: shell - uv sync --dev --extra airflow-providers --extra postgres + uv sync --all-extras --dev The *dev* dependency group includes development tools for code formatting, type checking, and testing. -The additional extras, *airflow-providers* and *postgres* install dependencies required for testing. If testing a specific Airflow version, the extras may be omitted, see the following section with more details. +The additional extras install dependencies required for testing. If testing a specific Airflow version, the extras may be omitted, see the following section with more details. + +*airflow-dbt-python* does not have a build system (yet!) so it needs to be installed separately: + +.. code-block:: shell + + uv pip install -e . Support for different versions of *Airflow* ------------------------------------------- @@ -96,7 +102,11 @@ The ``AIRFLOW_HOME`` environment variable has to be set to the same value used w The files ``airflow.cfg`` and ``airflow.db`` created as part of initializing the database can be safely deleted once not needed anymore. -Finally, some unit tests require Airflow provider packages. These are all provided by the *airflow-providers* extra. +Afterwards, ensure default connections are created by running: + +.. code-block:: shell + + uv run airflow connections create-default-connections Running tests ^^^^^^^^^^^^^ diff --git a/examples/readme_example_dbt_dag.py b/examples/readme_example_dbt_dag.py index 544a3b90..b710dbe8 100644 --- a/examples/readme_example_dbt_dag.py +++ b/examples/readme_example_dbt_dag.py @@ -28,7 +28,7 @@ ) as dag: dbt_test = DbtTestOperator( task_id="dbt_test", - selector_name="pre-run-tests", + selector="pre-run-tests", ) dbt_seed = DbtSeedOperator( diff --git a/tests/conftest.py b/tests/conftest.py index 34971dc3..10cf9f42 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -341,6 +341,23 @@ def model_files(dbt_project_dir): return paths +@pytest.fixture(scope="session") +def selector_file(model_files, dbt_project_dir): + """Create a selector file.""" + p = dbt_project_dir / "selectors.yml" + last_model = [str(m.stem) for m in model_files][-1] + + p.write_text(f"""\ +selectors: + - name: first_model + definition: + method: fqn + value: "{last_model}" + """) + + return p + + @pytest.fixture(scope="session") def sources_file(model_files, database): """Create test source file.""" diff --git a/tests/operators/test_dbt_build.py b/tests/operators/test_dbt_build.py index 6d58a04b..99df6c29 100644 --- a/tests/operators/test_dbt_build.py +++ b/tests/operators/test_dbt_build.py @@ -35,7 +35,7 @@ def test_dbt_build_mocked_all_args(): fail_fast=True, threads=3, exclude=["/path/to/model/to/exclude.sql"], - selector_name=["a-selector"], + selector="a-selector", state="/path/to/state/", singular=True, generic=True, @@ -62,7 +62,7 @@ def test_dbt_build_mocked_all_args(): "test_type:generic", ] assert config.exclude == ["/path/to/model/to/exclude.sql"] - assert config.selector_name == ["a-selector"] + assert config.selector == "a-selector" assert config.state == Path("/path/to/state/") assert config.singular is True assert config.generic is True diff --git a/tests/operators/test_dbt_compile.py b/tests/operators/test_dbt_compile.py index dc868401..6a66befd 100644 --- a/tests/operators/test_dbt_compile.py +++ b/tests/operators/test_dbt_compile.py @@ -25,7 +25,7 @@ def test_dbt_compile_mocked_all_args(): models=["/path/to/model1.sql", "/path/to/model2.sql"], threads=2, exclude=["/path/to/data/to/exclude.sql"], - selector_name=["a-selector"], + selector="a-selector", state="/path/to/state/", ) @@ -49,7 +49,7 @@ def test_dbt_compile_mocked_all_args(): "/path/to/model2.sql", ] assert config.exclude == ["/path/to/data/to/exclude.sql"] - assert config.selector_name == ["a-selector"] + assert config.selector == "a-selector" assert config.state == Path("/path/to/state/") @@ -198,9 +198,11 @@ def test_dbt_compile_uses_correct_argument_according_to_version(): models=["/path/to/model1.sql", "/path/to/model2.sql"], threads=2, exclude=["/path/to/data/to/exclude.sql"], - selector_name=["a-selector"], + selector_name="a-selector", state="/path/to/state/", ) assert op.select == ["/path/to/model1.sql", "/path/to/model2.sql"] assert getattr(op, "models", None) is None + assert op.selector == "a-selector" + assert getattr(op, "selector_name", None) is None diff --git a/tests/operators/test_dbt_run.py b/tests/operators/test_dbt_run.py index 2b0377ca..ef636396 100644 --- a/tests/operators/test_dbt_run.py +++ b/tests/operators/test_dbt_run.py @@ -35,7 +35,7 @@ def test_dbt_run_mocked_all_args(): fail_fast=True, threads=3, exclude=["/path/to/model/to/exclude.sql"], - selector_name=["a-selector"], + selector="a-selector", state="/path/to/state/", ) assert op.command == "run" @@ -54,7 +54,7 @@ def test_dbt_run_mocked_all_args(): assert config.threads == 3 assert config.select == ["/path/to/model.sql", "+/another/model.sql+2"] assert config.exclude == ["/path/to/model/to/exclude.sql"] - assert config.selector_name == ["a-selector"] + assert config.selector == "a-selector" assert config.state == Path("/path/to/state/") @@ -136,6 +136,38 @@ def test_dbt_run_models(profiles_file, dbt_project_file, model_files, logs_dir): ) +def test_dbt_run_models_with_first_model_selector( + profiles_file, dbt_project_file, model_files, selector_file, logs_dir +): + """Test execution of DbtRunOperator with a single model selector.""" + op = DbtRunOperator( + task_id="dbt_task", + project_dir=dbt_project_file.parent, + profiles_dir=profiles_file.parent, + selector="first_model", + do_xcom_push=True, + log_path=logs_dir, + log_level_file="info", + debug=True, + ) + + execution_results = op.execute({}) + run_result = execution_results["results"][0] + + assert run_result["status"] == RunStatus.Success + + log_file = logs_dir / "dbt.log" + assert log_file.exists() + + with open(log_file) as f: + logs = f.read() + + assert ( + "OK created view model public.model_4" in logs + or "OK created sql view model public.model_4" in logs + ) + + def test_dbt_run_models_with_env_vars( profiles_file_with_env, dbt_project_file, model_files, logs_dir, database ): @@ -334,12 +366,14 @@ def test_dbt_run_uses_correct_argument_according_to_version(): fail_fast=True, threads=3, exclude=["/path/to/model/to/exclude.sql"], - selector_name=["a-selector"], + selector_name="a-selector", state="/path/to/state/", ) assert op.select == ["/path/to/model.sql", "+/another/model.sql+2"] assert getattr(op, "models", None) is None + assert op.selector == "a-selector" + assert getattr(op, "selector_name", None) is None def test_dbt_run_models_with_airflow_connection( diff --git a/tests/operators/test_dbt_seed.py b/tests/operators/test_dbt_seed.py index 6db81f32..5c193a53 100644 --- a/tests/operators/test_dbt_seed.py +++ b/tests/operators/test_dbt_seed.py @@ -35,7 +35,7 @@ def test_dbt_seed_mocked_all_args(): show=True, threads=2, exclude=["/path/to/data/to/exclude.csv"], - selector_name=["a-selector"], + selector="a-selector", state="/path/to/state/", ) assert op.command == "seed" @@ -54,7 +54,7 @@ def test_dbt_seed_mocked_all_args(): assert config.select == ["/path/to/data.csv"] assert config.show is True assert config.exclude == ["/path/to/data/to/exclude.csv"] - assert config.selector_name == ["a-selector"] + assert config.selector == "a-selector" assert config.state == Path("/path/to/state/") diff --git a/tests/operators/test_dbt_snapshot.py b/tests/operators/test_dbt_snapshot.py index 4cfdaae8..cf74cbde 100644 --- a/tests/operators/test_dbt_snapshot.py +++ b/tests/operators/test_dbt_snapshot.py @@ -23,7 +23,7 @@ def test_dbt_snapshot_mocked_all_args(): select=["/path/to/models"], threads=2, exclude=["/path/to/data/to/exclude.sql"], - selector_name=["a-selector"], + selector="a-selector", state="/path/to/state/", ) assert op.command == "snapshot" @@ -40,7 +40,7 @@ def test_dbt_snapshot_mocked_all_args(): assert config.threads == 2 assert config.select == ["/path/to/models"] assert config.exclude == ["/path/to/data/to/exclude.sql"] - assert config.selector_name == ["a-selector"] + assert config.selector == "a-selector" assert config.state == Path("/path/to/state/") diff --git a/tests/operators/test_dbt_source.py b/tests/operators/test_dbt_source.py index 5a65726c..35b4c813 100644 --- a/tests/operators/test_dbt_source.py +++ b/tests/operators/test_dbt_source.py @@ -19,7 +19,7 @@ def test_dbt_source_mocked_all_args(): log_cache_events=True, select=["/path/to/models"], exclude=["/path/to/data/to/exclude.sql"], - selector_name=["a-selector"], + selector="a-selector", dbt_output="json", ) assert op.command == "source" @@ -35,7 +35,7 @@ def test_dbt_source_mocked_all_args(): assert config.log_cache_events is True assert config.select == ["/path/to/models"] assert config.exclude == ["/path/to/data/to/exclude.sql"] - assert config.selector_name == ["a-selector"] + assert config.selector == "a-selector" assert config.output == Output.JSON diff --git a/tests/operators/test_dbt_test.py b/tests/operators/test_dbt_test.py index 58ab7df6..d9020704 100644 --- a/tests/operators/test_dbt_test.py +++ b/tests/operators/test_dbt_test.py @@ -33,7 +33,7 @@ def test_dbt_test_configuration_all_args(): models=["/path/to/models"], threads=2, exclude=["/path/to/data/to/exclude.sql"], - selector_name=["a-selector"], + selector="a-selector", state="/path/to/state/", no_defer=True, fail_fast=True, @@ -60,7 +60,7 @@ def test_dbt_test_configuration_all_args(): "test_type:generic", ] assert config.exclude == ["/path/to/data/to/exclude.sql"] - assert config.selector_name == ["a-selector"] + assert config.selector == "a-selector" assert config.state == Path("/path/to/state/") assert config.no_defer is True @@ -239,10 +239,12 @@ def test_dbt_test_uses_correct_argument_according_to_version(): models=["/path/to/models"], threads=2, exclude=["/path/to/data/to/exclude.sql"], - selector_name=["a-selector"], + selector_name="a-selector", state="/path/to/state/", no_defer=True, ) assert op.select == ["/path/to/models"] assert getattr(op, "models", None) is None + assert op.selector == "a-selector" + assert getattr(op, "selector_name", None) is None