Support emitting Assets with Airflow 3

[tatiana]

Supports emitting Assets (Datasets) when using Cosmos with Airflow 3.

This implementation was tested in two ways:

1. By using airflow dags test

airflow dags test example_operators

We were able to observe via the logs that we're reaching the desired code:

[2025-04-28T13:30:44.272+0100] {runner.py:60} INFO - Trying to run dbtRunner with:
 ['run', '--models', 'stg_customers', '--project-dir', '/var/folders/td/522y78v91d1f5wgh67mj3p0m0000gn/T/tmp0wo7rsn5', '--profiles-dir', '/Users/tati/Code/cosmos-fresh/astronomer-cosmos/dev/dags/dbt/jaffle_shop', '--profile', 'default', '--target', 'dev']
 in /var/folders/td/522y78v91d1f5wgh67mj3p0m0000gn/T/tmp0wo7rsn5
12:30:44  Running with dbt=1.9.4
12:30:44  Registered adapter: postgres=1.9.0
12:30:44  Unable to do partial parsing because saved manifest not found. Starting full parse.
12:30:44  Found 5 models, 3 seeds, 18 data tests, 548 macros
12:30:44  
12:30:44  Concurrency: 4 threads (target='dev')
12:30:44  
12:30:45  1 of 1 START sql view model public.stg_customers ............................... [RUN]
12:30:45  1 of 1 OK created sql view model public.stg_customers .......................... [CREATE VIEW in 0.07s]
12:30:45  
12:30:45  Finished running 1 view model in 0 hours 0 minutes and 0.15 seconds (0.15s).
12:30:45  
12:30:45  Completed successfully
12:30:45  
12:30:45  Done. PASS=1 WARN=0 ERROR=0 SKIP=0 TOTAL=1
[2025-04-28T13:30:45.635+0100] {local.py:490} INFO - Inlets: []
[2025-04-28T13:30:45.635+0100] {local.py:491} INFO - Outlets: [Asset(name='postgres://localhost:5432/postgres.public.stg_customers', uri='postgres://localhost:5432/postgres.public.stg_customers', group='asset', extra={}, watchers=[])]

2. By using airflow standalone, triggering the DAG example_operators manually.

We also created a DAG that consumes the dataset created by example_operators, so we could confirm the DAG was being triggered via the generated dataset/alias:

from airflow import DAG
from airflow.sdk import Asset
from airflow.operators.empty import EmptyOperator


with DAG(
    "dataset_triggered_dag",
    description="A DAG that should be triggered via Dataset",
    schedule=[Asset("postgres://localhost:5432/postgres.public.stg_customers")],
) as dag:
    t1 = EmptyOperator(
        task_id="task_1",
    )
    t2 = EmptyOperator(
        task_id="task_2",
    )
    t1 >> t2

We were able to observe via the logs that we're reaching the desired code:

[2025-04-28, 13:18:30] INFO - Outlets: [Asset(name='postgres://localhost:5432/postgres.public.stg_customers', uri='postgres://localhost:5432/postgres.public.stg_customers', group='asset', extra={}, watchers=[])]: source="cosmos.operators.base"
[2025-04-28, 13:18:30] INFO - Assigning inlets/outlets with DatasetAlias in Airflow 3: source="cosmos.operators.local"

And we observed via Airflow UI the DAG dataset_triggered_dag being triggered.

Some screenshots:

During the process of implementing this feature, we identified a few limitations of Airflow 3.0.0 assets implementation, which were discussed with @uranusjr:

1. As of Airflow 3.0.0, we have to manually associate the AssetAlias to the self.outlets of the Operator
2. It is not possible to see outlets in the task instance UI
3. If the AssetAlias is created during task execution, the UI does not display the source DAG + event + triggered DAG
4. If we create the AssetAlias during task initialization, these are displayed incorrectly in the UI - and we don't want to clutter the user's UI with aliasses that may not link to any actual Assets:

Closes: #1635

[netlify]

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

Name	Link
🔨 Latest commit	d4d9c1c
🔍 Latest deploy log	https://app.netlify.com/sites/sunny-pastelito-5ecb04/deploys/68123f6227bf5d0008fff11f

[cloudflare-workers-and-pages]

Deploying astronomer-cosmos with    Cloudflare Pages

Latest commit:	d4d9c1c
Status:	⚡️  Build in progress...

View logs

[codecov]

Codecov Report

Attention: Patch coverage is 92.85714% with 1 line in your changes missing coverage. Please review.

Project coverage is 97.60%. Comparing base (a4e0805) to head (9ce5631).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
cosmos/operators/local.py	92.85%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1713      +/-   ##
==========================================
+ Coverage   97.58%   97.60%   +0.02%     
==========================================
  Files          83       83              
  Lines        5174     5180       +6     
==========================================
+ Hits         5049     5056       +7     
+ Misses        125      124       -1     

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

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Copilot]

Pull Request Overview

This PR adds support for emitting Assets (Datasets) when using Airflow 3 by updating test cases and adapting operator logic to work with the new Airflow APIs. Key changes include renaming tests to reflect Airflow version support, adding new integration tests for Airflow 3 behavior regarding AssetAlias, and modifying dataset handling (now using Assets instead of Datasets) in cosmos/operators/local.py.

Reviewed Changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
tests/operators/test_local.py	Updated tests to skip or adjust for Airflow 3 behavior and to raise errors as needed.
cosmos/operators/local.py	Refactored dataset handling to use Assets/AssetAlias and updated openlineage handling.

Comments suppressed due to low confidence (2)

tests/operators/test_local.py:563

• It appears the second log check is missing the 'assert' keyword, so the condition is not actually being verified. Please add 'assert' to ensure this log message is tested.

"Outlets: [Asset(name='postgres://0.0.0.0:5432/postgres/public.stg_customers', uri='postgres://0.0.0.0:5432/postgres/public.stg_customers'" in caplog.text

cosmos/operators/local.py:713

• [nitpick] Consider storing the AssetAlias instance in a local variable and reusing it (e.g. for both appending to self.outlets and as a dictionary key) to ensure consistency and avoid potential equality issues.

self.outlets.append(AssetAlias(dataset_alias_name))

[pankajkoti]

LGTM. Thanks for adding this support so quick. Have some questions in-line for my understanding and few minor suggestions if you'd like to consider.

Happy to merge the PR once CI passes.