Add structure to support multiple db for async operator execution

[pankajastro]

Pluggable Async Operator Interface

This PR enhances the initial async operator support in Cosmos,
as introduced in PR #1230. The changes decouple the DbtRunAirflowAsyncOperator
from BigQueryInsertJobOperator, making it more flexible and allowing
support for async operators with other data sources in the future

Introducing the DbtRunAirflowAsyncFactoryOperator class,
which dynamically selects the parent class containing
the async operator implementation based on dbt profile.

I’ve added a template for implementing the Databricks async operator.

ATM moved async operator-related code at path cosmos/operators/_async/, but open for suggestion
After discussing this with the team, I have moved async operator-related code at path cosmos/operators/_asynchronous/

Design principle

Introduced the DbtRunAirflowAsyncFactoryOperator class that uses a Factory Method design pattern combined with dynamic inheritance at runtime.

• Factory Method: The create_async_operator() method generates a specific async operator class based on the profile_config provided. This allows the operator to adapt to different types of async operator at runtime.
• Dynamic Inheritance: The class dynamically changes its base class (bases) to use the async operator class created in the factory method. This ensures the correct async class is used during execution.
• Execution: The execute() method calls the super().execute() to trigger the execution logic, but it dynamically uses the appropriate operator class for async behavior.

Class hierarchy

                                    BigQueryInsertJobOperator
                                               |
DbtRunLocalOperator  DbtRunAirflowAsyncBigqueryOperator DbtRunAirflowAsyncDatabricksOperator (inject these parent class at runtime)
              \              /
             DbtRunFactoryAirflowAsyncOperator
                         |
            DbtRunAirflowAsyncOperator

How to add a new async db operator

• Implement the operator at the path cosmos/operators/_asynchronous/
• The operator module should be in the format of: cosmos.operators._asynchronous.{profile_type}.{dbt_class}{_snake_case_to_camelcase(execution_mode)}{profile_type.capitalize()}Operator
• For details example, I have added a dummy implementation for Databricks

Example DAG

import os
from datetime import datetime
from pathlib import Path

from cosmos import DbtDag, ExecutionConfig, ExecutionMode, ProfileConfig, ProjectConfig, RenderConfig
from cosmos.profiles import GoogleCloudServiceAccountDictProfileMapping

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

profile_config = ProfileConfig(
    profile_name="default",
    target_name="dev",
    profile_mapping=GoogleCloudServiceAccountDictProfileMapping(
        conn_id="gcp_gs_conn", profile_args={"dataset": "release_17", "project": "astronomer-dag-authoring"}
    ),
)


# [START airflow_async_execution_mode_example]
simple_dag_async = DbtDag(
    # dbt/cosmos-specific parameters
    project_config=ProjectConfig(
        DBT_ROOT_PATH / "original_jaffle_shop",
    ),
    profile_config=profile_config,
    execution_config=ExecutionConfig(
        execution_mode=ExecutionMode.AIRFLOW_ASYNC,
    ),
    render_config=RenderConfig(
        select=["path:models"],
        # test_behavior=TestBehavior.NONE
    ),
    # normal dag parameters
    schedule_interval=None,
    start_date=datetime(2023, 1, 1),
    catchup=False,
    dag_id="simple_dag_async",
    tags=["simple"],
    operator_args={"full_refresh": True, "location": "northamerica-northeast1"},
)
# [END airflow_async_execution_mode_example]

Graph View

closes: #1238
closes: #1239

[netlify]

✅ Deploy Preview for sunny-pastelito-5ecb04 canceled.

Name	Link
🔨 Latest commit	fdd1f8c
🔍 Latest deploy log	https://app.netlify.com/sites/sunny-pastelito-5ecb04/deploys/679764f9a6a7d70008924597

[cloudflare-workers-and-pages]

Deploying astronomer-cosmos with    Cloudflare Pages

Latest commit:	fdd1f8c
Status:	✅  Deploy successful!
Preview URL:	https://97584574.astronomer-cosmos.pages.dev
Branch Preview URL:	https://async-execution-2.astronomer-cosmos.pages.dev

View logs

[codecov]

Codecov Report

Attention: Patch coverage is 95.00000% with 5 lines in your changes missing coverage. Please review.

Project coverage is 97.03%. Comparing base (2b99777) to head (fdd1f8c).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
cosmos/operators/_asynchronous/bigquery.py	90.56%	5 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1483      +/-   ##
==========================================
+ Coverage   96.62%   97.03%   +0.41%     
==========================================
  Files          74       77       +3     
  Lines        4413     4456      +43     
==========================================
+ Hits         4264     4324      +60     
+ Misses        149      132      -17     

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

[tatiana]

Hi @pankajastro , really exciting to see progress on this.

I understand the PR is still under draft, just giving some early feedback, as you requested it.

[tatiana]

Excellent work, @pankajastro. Great idea and implementation on dealing with the inheritance issues we've faced before.

Please, if possible, add a screenshot of the graph view in Airflow.