Can with DAG(..) be made thread local?

[csm10495]

If you happen to have multiple threads that are creating DAGs via with DAG(...) as dag:, but don't use the corresponding dag variable, and instead implicitly rely on the DagContext/get_current_dag(), you'll run into issues because the current DAG is global to the process, but is modified by each thread.

Here's a pseudo-y example case of what I mean:

...

def get_something_from_a_dag_object(*args, **kwargs):
   with DAG(*args, **kwargs) as dag:
       # time consuming operation

      # we forgot to explicitly give the dag to the task_group, so it used get_current_dag.
      @task_group(default_args={"retries": 3})
       def my_task_group():
           # stuff

      # do something with dag at this point.
...

t1 = Thread(target=get_dag_object, kargs=...)
t2 = Thread(target=get_dag_object, kargs=...)

t1.start()
t2.start()

t1.join()
t2.join()

The best solution is probably to use the dag object and pass it down as needed instead of relying on the current dag.

Though I'm kind of curious is there a reason that the DagContext can't be thread-local? Would making that object's storage thread-local allow the original pattern to work? (Could it break other things?)

[potiuk]

I don't think it makes sense to run dags in separate threads.

This:

# time consuming operation

Is something that should never happen. If you read best practices in our dags it's explicitly wrong to run time consuming operation when dag is being parsed. Don't.

You can do long running operations inside your "execute" methods of tasks, yes, but not when the Dags are parsed to establish the structure of the dag and let the Dag be serialized for scheduling.

Also threads in Python do something different than you think they do most likely. At least until Python 3.14 there is GIL (global interpreter lock) and Threads do not not help with running python code in parallel (they run concurrently but not in paralllel) - it only helps in case you have I/O operation and waiting for them by several threads essentially (which again is strictly "no-no" when it comes to best practices of Dag parsing).

So I see no particular reason why you would like to run parsing of Dags in separate threads as it gives you nothing if you follow good practice. Every Dag file is parsed in a separate process already to achieve paralllelism, so if you want to achieve parallelism, make several Dag files, import the common code from them and you are done. There is absolutely no need (and again - it makes no sense IMHO) to run Dag parsing in separate threads.

[csm10495]

I agree with you completely. There shouldn't be something time consuming in there.. but alas, that's how they wrote it. Even if it wasn't time consuming, there is still a chance for them to hit an issue here (though a smaller chance I guess). We're on older Python, so the GIL is still locking, though I'm guessing network or other IO calls within the DAG building process allow context switching.

They build the DAG inside a function inside a service outside of Airflow. The dag building has slow pieces and network calls inside (yes i agree this is bad). After they build the DAG, they parse it to figure out the flow the DAG would have then dump that info into another service.

Sort of like:

os.environ["_AIRFLOW__AS_LIBRARY"] = "1"
from airflow import DAG 

def load_dag_info_into_service(...):
    with DAG(...) as dag:
        ...
    # go through tasks in the dag, load them into the other service
    

The problem they recently hit was because the DAG-building-service uses a thread pool executor to distribute jobs (to that function ^) and they implicitly use the current dag, they sometimes get parsing issues because the current dag changes for 'the other thread' while 'this thread' is building.