Merge branch 'main' into support-static-index

.github/workflows/test.yml

@@ -38,11 +38,25 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11"]
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
         airflow-version: ["2.4", "2.5", "2.6", "2.7", "2.8", "2.9"]
         exclude:
           - python-version: "3.11"
             airflow-version: "2.4"
+          # Apache Airflow versions prior to 2.9.0 have not been tested with Python 3.12.
+          # Official support for Python 3.12 and the corresponding constraints.txt are available only for Apache Airflow >= 2.9.0.
+          # See: https://github.com/apache/airflow/tree/2.9.0?tab=readme-ov-file#requirements
+          # See: https://github.com/apache/airflow/tree/2.8.4?tab=readme-ov-file#requirements
+          - python-version: "3.12"
+            airflow-version: "2.4"
+          - python-version: "3.12"
+            airflow-version: "2.5"
+          - python-version: "3.12"
+            airflow-version: "2.6"
+          - python-version: "3.12"
+            airflow-version: "2.7"
+          - python-version: "3.12"
+            airflow-version: "2.8"
     steps:
       - uses: actions/checkout@v3
         with:

.pre-commit-config.yaml

@@ -54,7 +54,7 @@ repos:
           - --py37-plus
           - --keep-runtime-typing
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.4.5
+    rev: v0.4.7
     hooks:
       - id: ruff
         args:

cosmos/config.py

@@ -319,6 +319,18 @@ class ExecutionConfig:
     project_path: Path | None = field(init=False)
 
     def __post_init__(self, dbt_project_path: str | Path | None) -> None:
-        if self.invocation_mode and self.execution_mode != ExecutionMode.LOCAL:
-            raise CosmosValueError("ExecutionConfig.invocation_mode is only configurable for ExecutionMode.LOCAL.")
+        if self.invocation_mode and self.execution_mode not in (ExecutionMode.LOCAL, ExecutionMode.VIRTUALENV):
+            raise CosmosValueError(
+                "ExecutionConfig.invocation_mode is only configurable for ExecutionMode.LOCAL and ExecutionMode.VIRTUALENV."
+            )
+        if self.execution_mode == ExecutionMode.VIRTUALENV:
+            if self.invocation_mode == InvocationMode.DBT_RUNNER:
+                raise CosmosValueError(
+                    "InvocationMode.DBT_RUNNER has not been implemented for ExecutionMode.VIRTUALENV"
+                )
+            elif self.invocation_mode is None:
+                logger.debug(
+                    "Defaulting to InvocationMode.SUBPROCESS as it is the only supported invocation mode for ExecutionMode.VIRTUALENV"
+                )
+                self.invocation_mode = InvocationMode.SUBPROCESS
         self.project_path = Path(dbt_project_path) if dbt_project_path else None

cosmos/profiles/__init__.py

@@ -9,6 +9,7 @@
 from .bigquery.oauth import GoogleCloudOauthProfileMapping
 from .bigquery.service_account_file import GoogleCloudServiceAccountFileProfileMapping
 from .bigquery.service_account_keyfile_dict import GoogleCloudServiceAccountDictProfileMapping
+from .clickhouse.user_pass import ClickhouseUserPasswordProfileMapping
 from .databricks.token import DatabricksTokenProfileMapping
 from .exasol.user_pass import ExasolUserPasswordProfileMapping
 from .postgres.user_pass import PostgresUserPasswordProfileMapping
@@ -25,6 +26,7 @@
 
 profile_mappings: list[Type[BaseProfileMapping]] = [
     AthenaAccessKeyProfileMapping,
+    ClickhouseUserPasswordProfileMapping,
     GoogleCloudServiceAccountFileProfileMapping,
     GoogleCloudServiceAccountDictProfileMapping,
     GoogleCloudOauthProfileMapping,

cosmos/profiles/athena/access_key.py

@@ -66,9 +66,11 @@ def profile(self) -> dict[str, Any | None]:
             **self.profile_args,
             "aws_access_key_id": self.temporary_credentials.access_key,
             "aws_secret_access_key": self.get_env_var_format("aws_secret_access_key"),
-            "aws_session_token": self.get_env_var_format("aws_session_token"),
         }
 
+        if self.temporary_credentials.token:
+            profile["aws_session_token"] = self.get_env_var_format("aws_session_token")
+
         return self.filter_null(profile)
 
     @property

cosmos/profiles/bigquery/service_account_keyfile_dict.py

@@ -22,7 +22,6 @@ class GoogleCloudServiceAccountDictProfileMapping(BaseProfileMapping):
 
     required_fields = [
         "project",
-        "dataset",
         "keyfile_json",
     ]
 
@@ -45,12 +44,14 @@ def profile(self) -> dict[str, Any | None]:
         Even though the Airflow connection contains hard-coded Service account credentials,
         we generate a temporary file and the DBT profile uses it.
         """
-        return {
+        profile_dict = {
             **self.mapped_params,
             "threads": 1,
             **self.profile_args,
         }
 
+        return self.filter_null(profile_dict)
+
     @property
     def mock_profile(self) -> dict[str, Any | None]:
         "Generates mock profile. Defaults `threads` to 1."

cosmos/profiles/clickhouse/__init__.py

@@ -0,0 +1,5 @@
+"""Generic Airflow connection -> dbt profile mappings"""
+
+from .user_pass import ClickhouseUserPasswordProfileMapping
+
+__all__ = ["ClickhouseUserPasswordProfileMapping"]

cosmos/profiles/clickhouse/user_pass.py

@@ -0,0 +1,70 @@
+"""Maps Airflow Postgres connections using user + password authentication to dbt profiles."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ..base import BaseProfileMapping
+
+
+class ClickhouseUserPasswordProfileMapping(BaseProfileMapping):
+    """
+    Maps Airflow generic connections using user + password authentication to dbt Clickhouse profiles.
+    https://docs.getdbt.com/docs/core/connect-data-platform/clickhouse-setup
+    """
+
+    airflow_connection_type: str = "generic"
+    dbt_profile_type: str = "clickhouse"
+    default_port = 9000
+    is_community = True
+
+    required_fields = [
+        "host",
+        "login",
+        "schema",
+        "clickhouse",
+    ]
+    secret_fields = [
+        "password",
+    ]
+    airflow_param_mapping = {
+        "host": "host",
+        "login": "login",
+        "password": "password",
+        "port": "port",
+        "schema": "schema",
+        "clickhouse": "extra.clickhouse",
+    }
+
+    def _set_default_param(self, profile_dict: dict[str, Any]) -> dict[str, Any]:
+        if not profile_dict.get("driver"):
+            profile_dict["driver"] = "native"
+
+        if not profile_dict.get("port"):
+            profile_dict["port"] = self.default_port
+
+        if not profile_dict.get("secure"):
+            profile_dict["secure"] = False
+        return profile_dict
+
+    @property
+    def profile(self) -> dict[str, Any | None]:
+        """Gets profile. The password is stored in an environment variable."""
+        profile_dict = {
+            **self.mapped_params,
+            **self.profile_args,
+            # password should always get set as env var
+            "password": self.get_env_var_format("password"),
+        }
+
+        return self.filter_null(self._set_default_param(profile_dict))
+
+    @property
+    def mock_profile(self) -> dict[str, Any | None]:
+        """Gets mock profile."""
+
+        profile_dict = {
+            **super().mock_profile,
+        }
+
+        return self._set_default_param(profile_dict)

cosmos/settings.py

@@ -10,7 +10,7 @@
 # In MacOS users may want to set the envvar `TMPDIR` if they do not want the value of the temp directory to change
 DEFAULT_CACHE_DIR = Path(tempfile.gettempdir(), DEFAULT_COSMOS_CACHE_DIR_NAME)
 cache_dir = Path(conf.get("cosmos", "cache_dir", fallback=DEFAULT_CACHE_DIR) or DEFAULT_CACHE_DIR)
-enable_cache = conf.get("cosmos", "enable_cache", fallback=True)
+enable_cache = conf.getboolean("cosmos", "enable_cache", fallback=True)
 propagate_logs = conf.getboolean("cosmos", "propagate_logs", fallback=True)
 dbt_docs_dir = conf.get("cosmos", "dbt_docs_dir", fallback=None)
 dbt_docs_conn_id = conf.get("cosmos", "dbt_docs_conn_id", fallback=None)

docs/getting_started/execution-modes-local-conflicts.rst

@@ -10,24 +10,25 @@ If you find errors, we recommend users look into using `alternative execution mo
 
 In the following table, ``x`` represents combinations that lead to conflicts (vanilla ``apache-airflow`` and ``dbt-core`` packages):
 
-+---------------+-----+-----+-----+-----+-----+-----+-----+-----+
-| Airflow / DBT | 1.0 | 1.1 | 1.2 | 1.3 | 1.4 | 1.5 | 1.6 | 1.7 |
-+===============+=====+=====+=====+=====+=====+=====+=====+=====+
-| 2.2           |     |     |     | x   | x   | x   | x   | x   |
-+---------------+-----+-----+-----+-----+-----+-----+-----+-----+
-| 2.3           | x   | x   |     | x   | x   | x   | x   | x   |
-+---------------+-----+-----+-----+-----+-----+-----+-----+-----+
-| 2.4           | x   | x   | x   |     |     |     |     |     |
-+---------------+-----+-----+-----+-----+-----+-----+-----+-----+
-| 2.5           | x   | x   | x   |     |     |     |     |     |
-+---------------+-----+-----+-----+-----+-----+-----+-----+-----+
-| 2.6           | x   | x   | x   | x   | x   |     |     |     |
-+---------------+-----+-----+-----+-----+-----+-----+-----+-----+
-| 2.7           | x   | x   | x   | x   | x   |     |     |     |
-+---------------+-----+-----+-----+-----+-----+-----+-----+-----+
-| 2.8           | x   | x   | x   | x   | x   |     |  x  |  x  |
-+---------------+-----+-----+-----+-----+-----+-----+-----+-----+
-
++---------------+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+| Airflow / DBT | 1.0 | 1.1 | 1.2 | 1.3 | 1.4 | 1.5 | 1.6 | 1.7 | 1.8 |
++===============+=====+=====+=====+=====+=====+=====+=====+=====+=====+
+| 2.2           |     |     |     | x   | x   | x   | x   | x   | x   |
++---------------+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+| 2.3           | x   | x   |     | x   | x   | x   | x   | x   | X   |
++---------------+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+| 2.4           | x   | x   | x   |     |     |     |     |     |     |
++---------------+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+| 2.5           | x   | x   | x   |     |     |     |     |     |     |
++---------------+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+| 2.6           | x   | x   | x   | x   | x   |     |     |     |     |
++---------------+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+| 2.7           | x   | x   | x   | x   | x   |     |     |     |     |
++---------------+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+| 2.8           | x   | x   | x   | x   | x   |     |  x  |     |     |
++---------------+-----+-----+-----+-----+-----+-----+-----+-----+-----+
+| 2.9           | x   | x   | x   | x   | x   |     |     |     |     |
++---------------+-----+-----+-----+-----+-----+-----+-----+-----+-----+
 
 Examples of errors
 -----------------------------------
@@ -92,9 +93,11 @@ The table was created by running  `nox <https://nox.thea.codes/en/stable/>`__ wi
 
   @nox.session(python=["3.10"])
   @nox.parametrize(
-      "dbt_version", ["1.0", "1.1", "1.2", "1.3", "1.4", "1.5", "1.6", "1.7"]
+      "dbt_version", ["1.0", "1.1", "1.2", "1.3", "1.4", "1.5", "1.6", "1.7", "1.8"]
+  )
+  @nox.parametrize(
+      "airflow_version", ["2.2.4", "2.3", "2.4", "2.5", "2.6", "2.7", "2.8", "2.9"]
   )
-  @nox.parametrize("airflow_version", ["2.2.4", "2.3", "2.4", "2.5", "2.6", "2.7", "2.8"])
   def compatibility(session: nox.Session, airflow_version, dbt_version) -> None:
       """Run both unit and integration tests."""
       session.run(

docs/getting_started/execution-modes.rst

@@ -92,6 +92,7 @@ Some drawbacks of this approach:
 
 - It is slower than ``local`` because it creates a new Python virtual environment for each Cosmos dbt task run.
 - If dbt is unavailable in the Airflow scheduler, the default ``LoadMode.DBT_LS`` will not work. In this scenario, users must use a `parsing method <parsing-methods.html>`_  that does not rely on dbt, such as ``LoadMode.MANIFEST``.
+- Only ``InvocationMode.SUBPROCESS`` is supported currently, attempt to use ``InvocationMode.DBT_RUNNER`` will raise error.
 
 Example of how to use:
 

pyproject.toml

@@ -25,6 +25,7 @@ classifiers = [
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
 ]
 dependencies = [
     "aenum",
@@ -43,6 +44,7 @@ dependencies = [
 dbt-all = [
     "dbt-athena",
     "dbt-bigquery",
+    "dbt-clickhouse",
     "dbt-databricks",
     "dbt-exasol",
     "dbt-postgres",
@@ -53,6 +55,7 @@ dbt-all = [
 ]
 dbt-athena = ["dbt-athena-community", "apache-airflow-providers-amazon>=8.0.0"]
 dbt-bigquery = ["dbt-bigquery"]
+dbt-clickhouse = ["dbt-clickhouse"]
 dbt-databricks = ["dbt-databricks"]
 dbt-exasol = ["dbt-exasol"]
 dbt-postgres = ["dbt-postgres"]
@@ -138,7 +141,7 @@ dependencies = [
 pre-install-commands = ["sh scripts/test/pre-install-airflow.sh {matrix:airflow} {matrix:python}"]
 
 [[tool.hatch.envs.tests.matrix]]
-python = ["3.8", "3.9", "3.10", "3.11"]
+python = ["3.8", "3.9", "3.10", "3.11", "3.12"]
 airflow = ["2.4", "2.5", "2.6", "2.7", "2.8", "2.9"]
 
 [tool.hatch.envs.tests.overrides]

tests/operators/test_base.py

@@ -1,3 +1,4 @@
+import sys
 from unittest.mock import patch
 
 import pytest
@@ -14,6 +15,10 @@
 )
 
 
+@pytest.mark.skipif(
+    (sys.version_info.major, sys.version_info.minor) == (3, 12),
+    reason="The error message for the abstract class instantiation seems to have changed between Python 3.11 and 3.12",
+)
 def test_dbt_base_operator_is_abstract():
     """Tests that the abstract base operator cannot be instantiated since the base_cmd is not defined."""
     expected_error = (
@@ -23,6 +28,20 @@ def test_dbt_base_operator_is_abstract():
         AbstractDbtBaseOperator()
 
 
+@pytest.mark.skipif(
+    (sys.version_info.major, sys.version_info.minor) != (3, 12),
+    reason="The error message for the abstract class instantiation seems to have changed between Python 3.11 and 3.12",
+)
+def test_dbt_base_operator_is_abstract_py12():
+    """Tests that the abstract base operator cannot be instantiated since the base_cmd is not defined."""
+    expected_error = (
+        "Can't instantiate abstract class AbstractDbtBaseOperator without an implementation for abstract methods "
+        "'base_cmd', 'build_and_run_cmd'"
+    )
+    with pytest.raises(TypeError, match=expected_error):
+        AbstractDbtBaseOperator()
+
+
 @pytest.mark.parametrize("cmd_flags", [["--some-flag"], []])
 @patch("cosmos.operators.base.AbstractDbtBaseOperator.build_and_run_cmd")
 def test_dbt_base_operator_execute(mock_build_and_run_cmd, cmd_flags, monkeypatch):