Is @DAG/@task decorators compatible with dynamically generation?

[makrushin-evgenii]

Apache Airflow version

2.2.3 (latest released)

What happened

from datetime import datetime
from airflow.decorators import dag, task


for n in range(1, 4):
    dag_id = "test_dag_{}".format(str(n))

    @dag(dag_id=dag_id, schedule_interval=None, start_date=datetime(2022, 2, 1), catchup=False)
    def test_dag():
        @task
        def test_task(x):
            print("dag_id={}, x={}".format(dag_id, x))

        test_task(dag_id)

    globals()[dag_id] = test_dag()

Running the DAG above, i retrieve the logs:

[2022-02-02, 16:44:26 UTC] {taskinstance.py:1426} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=***
AIRFLOW_CTX_DAG_ID=test_dag_1
AIRFLOW_CTX_TASK_ID=test_task
AIRFLOW_CTX_EXECUTION_DATE=2022-02-02T16:32:55.747008+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2022-02-02T16:32:55.747008+00:00
[2022-02-02, 16:44:26 UTC] {logging_mixin.py:109} INFO - dag_id=test_dag_3, x=test_dag_1

[2022-02-02, 16:51:49 UTC] {taskinstance.py:1426} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=***
AIRFLOW_CTX_DAG_ID=test_dag_2
AIRFLOW_CTX_TASK_ID=test_task
AIRFLOW_CTX_EXECUTION_DATE=2022-02-02T16:33:08.308699+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2022-02-02T16:33:08.308699+00:00
[2022-02-02, 16:51:49 UTC] {logging_mixin.py:109} INFO - dag_id=test_dag_3, x=test_dag_2

1. During the DAG generation, dag_id variable tooks different values
2. During the execution test_task method, dag_id variable tooks the single value "test_dag_3"
3. During the execution test_task method, dag_id variable tooks different values when passed through x param

What you expected to happen

The variable takes the same value during DAG generation and methods execution:

[2022-02-02, 16:44:26 UTC] {taskinstance.py:1426} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_ID=test_dag_1
[2022-02-02, 16:44:26 UTC] {logging_mixin.py:109} INFO - dag_id=test_dag_1, x=test_dag_1

[2022-02-02, 16:51:49 UTC] {taskinstance.py:1426} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_ID=test_dag_2
[2022-02-02, 16:51:49 UTC] {logging_mixin.py:109} INFO - dag_id=test_dag_2, x=test_dag_2

(logs edited to show expected result)

How to reproduce

Run the DAG above

Operating System

macOS Monterey 12.1

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!

[potiuk]

This is not an issue - this is a discussion. I believe it is pretty expectedw way how decorators work, but I wonder what others have to say on thtat one.

[makrushin-evgenii]

Can you explain why it is expected? Looks like a difference between lazy and strict evaluation, but:

1. it's clear why AIRFLOW_CTX_DAG_ID can differ from variable value
2. it's not clear why the single variable at the single moment takes different values depending on the method of call: directly (dag_id=) or passed through method param (x=)

[potiuk]

I did not look in detail - but maybe others will be able to chime in - especially those involved in the decorator's work?

[potiuk]

cc: @uranusjr @ashb @josh-fell -> I think you know some of the intrinsic details on how decorators work and why the behaviour there (which indeed seems surprising so maybe we should better document it). Or maybe we can fix it to behave more "intuitively".

[potiuk]

or maybe even @turbaszek who was one of the original authors and reviewers :)

[ashb]

This isn't anything to do with the TaskFlow directly, it's how python scoping works with for-loops, coupled with now DAGs are parsed.

• When parsing the DAG file (which happens up front) test_task(dag_id) will be called with 1..3 as the values
• At this point, the value of the dag_id variable is left at test_dag_3
• Then Airflow says "I want to run dag X", pulls the dag out of the module.

So when the task function is executed, the current/latest value of dag_id variable is the one you get.

The way to do this and get the behaviour you want is to use dag from the context:

    @dag(dag_id=dag_id, schedule_interval=None, start_date=datetime(2022, 2, 1), catchup=False)
    def test_dag():
        @task
        def test_task(x, dag):
            print("dag_id={}, x={}".format(dag.dag_id, x))

        test_task(dag_id)

[potiuk]

@makrushin-evgenii - maybe you would like to add a PR to our docs to add what @ashb explained :) ? I think this would be a nice contribution and it's easy: "Suggest a change on this page" button at the right- bottom of the page? You can easily become one of the > 1900 contributors this way :D

[uranusjr]

Also note that this property is not limited to Airflow’s DAG parsing, but to many situations in general. For example:

functions = []

for i in range(2):
    def print_me():
        print(i)

    functions.append(print_me)

for f in functions:
    f()

You may expect this to print 0 and 1, but in fact it will print 1 two times instead. This is often coined as late binding.

[makrushin-evgenii]

@potiuk, open docs PR with this method of DAGs generation with link to @uranusjr explanation of late binding behaviour