Support customizing how dbt nodes are converted to Airflow

[tatiana]

Description

This PR aims to solve the following current limitations of Cosmos 1.0:

• If you want to pass arguments to just some tasks (e.g., on_warning_callback to test nodes), Cosmos has to add it to the main interface and explicitly pass it down to just those tasks
• If a user wants to subclass one of the Cosmos operators and use that instead, they can't
• If a user wants more granular customization over how each task is rendered, they can't use Cosmos

It does this by introducing the parameter node_converters to RenderConfig, which allows users to define a custom function to convert a DbtNode into nothing or an Airflow resource (Operator or TaskGroup instances).

Example of the feature

import os
from datetime import datetime
from pathlib import Path

from airflow.operators.dummy import DummyOperator
from airflow.models.dag import DAG
from airflow.utils.task_group import TaskGroup

from cosmos import DbtDag, ProjectConfig, ProfileConfig, RenderConfig
from cosmos.constants import DbtResourceType
from cosmos.dbt.graph import DbtNode

DEFAULT_DBT_ROOT_PATH = Path(__file__).parent / "dbt"
DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))

os.environ["DBT_SQLITE_PATH"] = str(DEFAULT_DBT_ROOT_PATH / "simple")


profile_config = ProfileConfig(
    profile_name="simple",
    target_name="dev",
    profiles_yml_filepath=(DBT_ROOT_PATH / "simple/profiles.yml"),
)


# [START custom_dbt_nodes]
def convert_source(dag: DAG, task_group: TaskGroup, node: DbtNode, **kwargs):
    return DummyOperator(dag=dag, task_group=task_group, task_id=f"{node.name}_source")


def convert_exposure(dag: DAG, task_group: TaskGroup, node: DbtNode, **kwargs):
    return DummyOperator(dag=dag, task_group=task_group, task_id=f"{node.name}_exposure")


render_config = RenderConfig(
    node_converters={DbtResourceType.SOURCE: convert_source, DbtResourceType("exposure"): convert_exposure}
)


example_cosmos_sources = DbtDag(
    # dbt/cosmos-specific parameters
    project_config=ProjectConfig(
        DBT_ROOT_PATH / "simple",
    ),
    profile_config=profile_config,
    render_config=render_config,
    operator_args={"append_env": True},
    # normal dag parameters
    schedule_interval="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False,
    dag_id="example_cosmos_sources",
)
# [END custom_dbt_nodes]

Is now rendered in Airflow:

Before this change, there was no way for users to describe how to render source or other unsupported dbt resources.

Related Issue(s)

Closes: #427
Closes: #477

Breaking Change?

No

Checklist

• I have made corresponding changes to the documentation (if required)
• I have added tests that prove my fix is effective or that my feature works

Feedback addressed:

• rename dbt_resource_converter to node_converters
• support passing items not previously supported in the enum DbtResourceType
• write docs and say this feature is experimental

[netlify]

👷 Deploy Preview for amazing-pothos-a3bca0 processing.

Name	Link
🔨 Latest commit	945f62f
🔍 Latest deploy log	https://app.netlify.com/sites/amazing-pothos-a3bca0/deploys/652914cc1ddb330008a23cf9

[codecov]

Codecov Report

Attention: 1 lines in your changes are missing coverage. Please review.

Comparison is base (9460312) 93.01% compared to head (945f62f) 93.04%.

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #503      +/-   ##
==========================================
+ Coverage   93.01%   93.04%   +0.02%     
==========================================
  Files          51       51              
  Lines        2019     2041      +22     
==========================================
+ Hits         1878     1899      +21     
- Misses        141      142       +1     

Files	Coverage Δ
cosmos/config.py	92.94% <100.00%> (+0.08%)	⬆️
cosmos/constants.py	100.00% <100.00%> (ø)
cosmos/converter.py	97.10% <100.00%> (+0.04%)	⬆️
cosmos/dbt/graph.py	98.79% <100.00%> (+<0.01%)	⬆️
cosmos/airflow/graph.py	98.75% <96.15%> (-1.25%)	⬇️

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.