Release 1.2.2

.github/workflows/test.yml

@@ -121,8 +121,9 @@ jobs:
         env:
           AIRFLOW_HOME: /home/runner/work/astronomer-cosmos/astronomer-cosmos/
           AIRFLOW_CONN_AIRFLOW_DB: postgres://postgres:[email protected]:5432/postgres
-          PYTHONPATH: /home/runner/work/astronomer-cosmos/astronomer-cosmos/:$PYTHONPATH
           AIRFLOW_CONN_DATABRICKS_DEFAULT: ${{ secrets.AIRFLOW_CONN_DATABRICKS_DEFAULT }}
+          AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT: 90.0
+          PYTHONPATH: /home/runner/work/astronomer-cosmos/astronomer-cosmos/:$PYTHONPATH
           DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
           DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
           DATABRICKS_WAREHOUSE_ID: ${{ secrets.DATABRICKS_WAREHOUSE_ID }}
@@ -192,6 +193,7 @@ jobs:
           AIRFLOW_CONN_AIRFLOW_DB: postgres://postgres:[email protected]:5432/postgres
           PYTHONPATH: /home/runner/work/astronomer-cosmos/astronomer-cosmos/:$PYTHONPATH
           AIRFLOW_CONN_DATABRICKS_DEFAULT: ${{ secrets.AIRFLOW_CONN_DATABRICKS_DEFAULT }}
+          AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT: 90.0
           DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}
           DATABRICKS_TOKEN: ${{ secrets.DATABRICKS_TOKEN }}
           DATABRICKS_WAREHOUSE_ID: ${{ secrets.DATABRICKS_WAREHOUSE_ID }}
@@ -256,6 +258,7 @@ jobs:
         env:
           AIRFLOW_HOME: /home/runner/work/astronomer-cosmos/astronomer-cosmos/
           AIRFLOW_CONN_AIRFLOW_DB: postgres://postgres:[email protected]:5432/postgres
+          AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT: 90.0
           PYTHONPATH: /home/runner/work/astronomer-cosmos/astronomer-cosmos/:$PYTHONPATH
           AIRFLOW_CONN_DATABRICKS_DEFAULT: ${{ secrets.AIRFLOW_CONN_DATABRICKS_DEFAULT }}
           DATABRICKS_HOST: ${{ secrets.DATABRICKS_HOST }}

.pre-commit-config.yaml

@@ -53,13 +53,13 @@ repos:
           - --py37-plus
           - --keep-runtime-typing
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.1.1
+    rev: v0.1.3
     hooks:
       - id: ruff
         args:
           - --fix
   - repo: https://github.com/psf/black
-    rev: 23.10.0
+    rev: 23.10.1
     hooks:
       - id: black
         args: [ "--config", "./pyproject.toml" ]
@@ -76,6 +76,12 @@ repos:
         name: mypy-python
         additional_dependencies: [types-PyYAML, types-attrs, attrs, types-requests, types-python-dateutil, apache-airflow]
         files: ^cosmos
+  - repo: https://github.com/pycqa/flake8
+    rev: 6.1.0
+    hooks:
+      - id: flake8
+        entry: pflake8
+        additional_dependencies: [pyproject-flake8]
 
 ci:
   autofix_commit_msg: 🎨 [pre-commit.ci] Auto format from pre-commit.com hooks

CHANGELOG.rst

@@ -1,6 +1,22 @@
 Changelog
 =========
 
+1.2.2 (2023-11-06)
+------------------
+
+Bug fixes
+
+* Support ``ProjectConfig.dbt_project_path = None`` & different paths for Rendering and Execution by @MrBones757 in #634
+* Fix adding test nodes to DAGs built using ``LoadMethod.DBT_MANIFEST`` and ``LoadMethod.CUSTOM`` by @edgga in #615
+
+Others
+
+* Add pre-commit hook for McCabe max complexity check and fix errors by @jbandoro in #629
+* Update contributing docs for running integration tests by @jbandoro in #638
+* Fix CI issue running integration tests by @tatiana in #640 and #644
+* pre-commit updates in #637
+
+
 1.2.1 (2023-10-25)
 ------------------
 

cosmos/__init__.py

@@ -5,7 +5,7 @@
 
 Contains dags, task groups, and operators.
 """
-__version__ = "1.2.1"
+__version__ = "1.2.2"
 
 from cosmos.airflow.dag import DbtDag
 from cosmos.airflow.task_group import DbtTaskGroup

cosmos/config.py

@@ -4,7 +4,7 @@
 
 import contextlib
 import tempfile
-from dataclasses import dataclass, field
+from dataclasses import InitVar, dataclass, field
 from pathlib import Path
 from typing import Any, Iterator, Callable
 
@@ -31,6 +31,9 @@ class RenderConfig:
     :param select: A list of dbt select arguments (e.g. 'config.materialized:incremental')
     :param exclude: A list of dbt exclude arguments (e.g. 'tag:nightly')
     :param dbt_deps: Configure to run dbt deps when using dbt ls for dag parsing
+    :param node_converters: a dictionary mapping a ``DbtResourceType`` into a callable. Users can control how to render dbt nodes in Airflow. Only supported when using ``load_method=LoadMode.DBT_MANIFEST`` or ``LoadMode.DBT_LS``.
+    :param dbt_executable_path: The path to the dbt executable for dag generation. Defaults to dbt if available on the path. Mutually Exclusive with ProjectConfig.dbt_project_path
+    :param dbt_project_path Configures the DBT project location accessible on the airflow controller for DAG rendering - Required when using ``load_method=LoadMode.DBT_LS`` or ``load_method=LoadMode.CUSTOM``
     """
 
     emit_datasets: bool = True
@@ -40,6 +43,13 @@ class RenderConfig:
     exclude: list[str] = field(default_factory=list)
     dbt_deps: bool = True
     node_converters: dict[DbtResourceType, Callable[..., Any]] | None = None
+    dbt_executable_path: str | Path = get_system_dbt()
+    dbt_project_path: InitVar[str | Path | None] = None
+
+    project_path: Path | None = field(init=False)
+
+    def __post_init__(self, dbt_project_path: str | Path | None) -> None:
+        self.project_path = Path(dbt_project_path) if dbt_project_path else None
 
 
 class ProjectConfig:
@@ -72,11 +82,13 @@ def __init__(
         manifest_path: str | Path | None = None,
         project_name: str | None = None,
     ):
+        # Since we allow dbt_project_path to be defined in ExecutionConfig and RenderConfig
+        #   dbt_project_path may not always be defined here.
+        # We do, however, still require that both manifest_path and project_name be defined, or neither be defined.
         if not dbt_project_path:
-            if not manifest_path or not project_name:
+            if manifest_path and not project_name or project_name and not manifest_path:
                 raise CosmosValueError(
-                    "ProjectConfig requires dbt_project_path and/or manifest_path to be defined."
-                    " If only manifest_path is defined, project_name must also be defined."
+                    "If ProjectConfig.dbt_project_path is not defined, ProjectConfig.manifest_path and ProjectConfig.project_name must be defined together, or both left undefined."
                 )
         if project_name:
             self.project_name = project_name
@@ -210,10 +222,17 @@ class ExecutionConfig:
 
     :param execution_mode: The execution mode for dbt. Defaults to local
     :param test_indirect_selection: The mode to configure the test behavior when performing indirect selection.
-    :param dbt_executable_path: The path to the dbt executable. Defaults to dbt if
-    available on the path.
+    :param dbt_executable_path: The path to the dbt executable for runtime execution. Defaults to dbt if available on the path.
+    :param dbt_project_path Configures the DBT project location accessible at runtime for dag execution. This is the project path in a docker container for ExecutionMode.DOCKER or ExecutionMode.KUBERNETES. Mutually Exclusive with ProjectConfig.dbt_project_path
     """
 
     execution_mode: ExecutionMode = ExecutionMode.LOCAL
     test_indirect_selection: TestIndirectSelection = TestIndirectSelection.EAGER
     dbt_executable_path: str | Path = get_system_dbt()
+
+    dbt_project_path: InitVar[str | Path | None] = None
+
+    project_path: Path | None = field(init=False)
+
+    def __post_init__(self, dbt_project_path: str | Path | None) -> None:
+        self.project_path = Path(dbt_project_path) if dbt_project_path else None

cosmos/converter.py

@@ -16,7 +16,6 @@
 from cosmos.exceptions import CosmosValueError
 from cosmos.log import get_logger
 
-
 logger = get_logger(__name__)
 
 
@@ -92,8 +91,8 @@ def __init__(
         self,
         project_config: ProjectConfig,
         profile_config: ProfileConfig,
-        execution_config: ExecutionConfig = ExecutionConfig(),
-        render_config: RenderConfig = RenderConfig(),
+        execution_config: ExecutionConfig | None = None,
+        render_config: RenderConfig | None = None,
         dag: DAG | None = None,
         task_group: TaskGroup | None = None,
         operator_args: dict[str, Any] | None = None,
@@ -103,19 +102,37 @@ def __init__(
     ) -> None:
         project_config.validate_project()
 
-        emit_datasets = render_config.emit_datasets
-        test_behavior = render_config.test_behavior
-        select = render_config.select
-        exclude = render_config.exclude
-        dbt_deps = render_config.dbt_deps
-        execution_mode = execution_config.execution_mode
-        test_indirect_selection = execution_config.test_indirect_selection
-        load_mode = render_config.load_method
-        dbt_executable_path = execution_config.dbt_executable_path
-        node_converters = render_config.node_converters
-
-        if not project_config.dbt_project_path:
-            raise CosmosValueError("A Project Path in ProjectConfig is required for generating a Task Operators.")
+        if not execution_config:
+            execution_config = ExecutionConfig()
+        if not render_config:
+            render_config = RenderConfig()
+
+        # Since we now support both project_config.dbt_project_path, render_config.project_path and execution_config.project_path
+        # We need to ensure that only one interface is being used.
+        if project_config.dbt_project_path and (render_config.project_path or execution_config.project_path):
+            raise CosmosValueError(
+                "ProjectConfig.dbt_project_path is mutually exclusive with RenderConfig.dbt_project_path and ExecutionConfig.dbt_project_path."
+                + "If using RenderConfig.dbt_project_path or ExecutionConfig.dbt_project_path, ProjectConfig.dbt_project_path should be None"
+            )
+
+        # If we are using the old interface, we should migrate it to the new interface
+        # This is safe to do now since we have validated which config interface we're using
+        if project_config.dbt_project_path:
+            render_config.project_path = project_config.dbt_project_path
+            execution_config.project_path = project_config.dbt_project_path
+
+        # At this point, execution_config.project_path should always be non-null
+        if not execution_config.project_path:
+            raise CosmosValueError(
+                "ExecutionConfig.dbt_project_path is required for the execution of dbt tasks in all execution modes."
+            )
+
+        # We now have a guaranteed execution_config.project_path, but still need to process render_config.project_path
+        # We require render_config.project_path when we dont have a manifest
+        if not project_config.manifest_path and not render_config.project_path:
+            raise CosmosValueError(
+                "RenderConfig.dbt_project_path is required for rendering an airflow DAG from a DBT Graph if no manifest is provided."
+            )
 
         profile_args = {}
         if profile_config.profile_mapping:
@@ -136,36 +153,34 @@ def __init__(
         # We may want to consider defaulting this value in our actual ProjceConfig class?
         dbt_graph = DbtGraph(
             project=project_config,
-            exclude=exclude,
-            select=select,
-            dbt_cmd=dbt_executable_path,
+            render_config=render_config,
+            execution_config=execution_config,
+            dbt_cmd=render_config.dbt_executable_path,
             profile_config=profile_config,
             operator_args=operator_args,
-            dbt_deps=dbt_deps,
         )
-        dbt_graph.load(method=load_mode, execution_mode=execution_mode)
+        dbt_graph.load(method=render_config.load_method, execution_mode=execution_config.execution_mode)
 
         task_args = {
             **operator_args,
-            # the following args may be only needed for local / venv:
-            "project_dir": project_config.dbt_project_path,
+            "project_dir": execution_config.project_path,
             "profile_config": profile_config,
-            "emit_datasets": emit_datasets,
+            "emit_datasets": render_config.emit_datasets,
         }
-        if dbt_executable_path:
-            task_args["dbt_executable_path"] = dbt_executable_path
+        if execution_config.dbt_executable_path:
+            task_args["dbt_executable_path"] = execution_config.dbt_executable_path
 
-        validate_arguments(select, exclude, profile_args, task_args)
+        validate_arguments(render_config.select, render_config.exclude, profile_args, task_args)
 
         build_airflow_graph(
             nodes=dbt_graph.filtered_nodes,
             dag=dag or (task_group and task_group.dag),
             task_group=task_group,
-            execution_mode=execution_mode,
+            execution_mode=execution_config.execution_mode,
             task_args=task_args,
-            test_behavior=test_behavior,
-            test_indirect_selection=test_indirect_selection,
+            test_behavior=render_config.test_behavior,
+            test_indirect_selection=execution_config.test_indirect_selection,
             dbt_project_name=project_config.project_name,
             on_warning_callback=on_warning_callback,
-            node_converters=node_converters,
+            node_converters=render_config.node_converters,
         )