Using TriggerDagRunOperator to run dags with names retrieved from XCom

[fam-epam]

Airflow version: 2.1.4

I would like to trigger a dag with the name stored in XCom.
I tried the following:

re_triggered = TriggerDagRunOperator( task_id='re_triggered', trigger_dag_id="{{ ti.xcom_pull('redaction_dag', key='redaction_pipeline') }}", execution_date=datetime.now(timezone.utc), reset_dag_run=True, wait_for_completion=False, dag=dag )

I got the following error:

DAG "{{ ti.xcom_pull('redaction_dag', key='redaction_pipeline') }}" seems to be missing from DagBag.

1. Is it possible to use a "trigger_dag_id" retrieved from XCom?

Thank you!

[potiuk]

Depends which context you are running it with. You need to show the whole DAG you have because it does not look you used it properly ('trigger_dag_id' is a templated field so it should work as you expected and it has been in templated_fields since 2018 or so as I checked). You must have been doing something wrong in your DAG definition.

[fam-epam]

Thanks for looking into it.

I pasted below a simplified (but working) version of the dag together with the full error I get:

import code
import json
import logging
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Tuple

from airflow import DAG
from airflow.models import taskinstance
from airflow.operators.python import PythonOperator
from airflow.operators.trigger_dagrun import TriggerDagRunOperator
import requests

NUMBER_STREAMS = 3


LOGGER = logging.getLogger(__name__)
logging.basicConfig(
    level="INFO",
    format="%(asctime)s %(name)s %(levelname)s %(message)s",
)

def end_task():
    LOGGER.info("END")

def start_task():
    LOGGER.info("START")


def start_redaction_process(session, task_params, **kwargs):
    # ti = kwargs["task_instance"]
    # ti.xcom_push(key="airflow_pipeline_name", value="DPTestRedaction")
    return "DPTestRedaction"
    
    
def test(session, task_params, dag_name, **kwargs):

    # ti = kwargs["task_instance"]
    
    LOGGER.info(
        f"TEST, "
        f"dag_name={dag_name}, "
    )

    
with DAG(
    dag_id="DPProcessingRedaction",
    default_args={
        "depends_on_past": False,
        "email": ["airflow@example.com"],
        "email_on_failure": False,
        "email_on_retry": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=5),
    },
    description="Executing the redaction process",
    start_date=datetime(2022, 1, 1),
    catchup=False,
    tags=["DPRedactionL1"],
    render_template_as_native_obj=True,
) as dag:

    s = requests.Session()

    start_task = PythonOperator(
        task_id="start_task",
        python_callable=start_task,
        provide_context=True,
        do_xcom_push=True,
    )

    end_task = PythonOperator(
        task_id="end_task",
        python_callable=end_task,
        provide_context=True,
        do_xcom_push=True,
    )

    tasks = {}
    for id in range(NUMBER_STREAMS):
        tasks[id] = {}

        task_params = {
            "matches_list_index": id,
            "matches_list_task_name": "get_redaction_matches"
        }
        tasks[id]["start_redaction_process"] = PythonOperator(
            task_id=f"start_redaction_process_{id}",
            python_callable=start_redaction_process,
            provide_context=True,
            do_xcom_push=True,
            op_args=[s, task_params],
        )
        
        task_params = {
            "matches_list_index": id,
            "matches_list_task_name": "get_redaction_matches"
        }
        tasks[id]["test"] = PythonOperator(
            task_id=f"test_{id}",
            python_callable=test,
            provide_context=True,
            do_xcom_push=True,
            op_args=[s, task_params, "{{ ti.xcom_pull('start_redaction_process_" + str(id) + "', key='return_value') }}"]
        )

        timezone_offset = 0
        tzinfo = timezone(timedelta(hours=timezone_offset))

        dag_id_to_trigger = "{{ ti.xcom_pull('start_redaction_process_" + str(id) + "', key='return_value') }}"
        tasks[id]["rerun_existing_pipeline"] = TriggerDagRunOperator(
            task_id=f'rerun_existing_pipeline_{id}',
            trigger_dag_id=dag_id_to_trigger,
            #trigger_dag_id="DPTestRedaction",
            execution_date=datetime.now(timezone.utc),
            reset_dag_run=True,
            wait_for_completion=False,
            poke_interval=10,
            dag=dag
        )
       


        (
            start_task 
            >> tasks[id]["start_redaction_process"]
            >> tasks[id]["test"]
            >> tasks[id]["rerun_existing_pipeline"]
            >> end_task
        )

This is the full error I get:

Something bad has happened.
Please consider letting us know by creating a [bug report using GitHub](https://github.com/apache/airflow/issues/new/choose).

Python version: 3.8.10
Airflow version: 2.1.4
Node: EPROBUCW005F.localdomain
-------------------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/airflow/www/auth.py", line 49, in decorated
    return func(*args, **kwargs)
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/airflow/www/decorators.py", line 97, in view_func
    return f(*args, **kwargs)
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/airflow/www/decorators.py", line 60, in wrapper
    return f(*args, **kwargs)
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/airflow/www/views.py", line 2027, in tree
    dag = current_app.dag_bag.get_dag(dag_id)
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/airflow/models/dagbag.py", line 186, in get_dag
    self._add_dag_from_db(dag_id=dag_id, session=session)
  File "/home/flavius/venv_airflow_2.1.4/lib/python3.8/site-packages/airflow/models/dagbag.py", line 258, in _add_dag_from_db
    raise SerializedDagNotFound(f"DAG '{dag_id}' not found in serialized_dag table")
airflow.exceptions.SerializedDagNotFound: DAG '{{ ti.xcom_pull('start_redaction_process_0', key='return_value') }}' not found in serialized_dag table

[fam-epam]

@potiuk have you had a chance to take a look over the code I sent? I wonder if there is any solution for that.
Thanks!

[potiuk]

No. I've been pretty busy recently. I will definitely come back to it at some point of time though, when I handle the current pile of things. That's the natue of open-source - we help when we have time to spare - mostly in our free time.

[fam-epam]

Thank you for your kind answer. I would really appreciate your help whenever you can find some time to look over the code.

[potiuk]

Seems your error is from UI - so likely this is or WAS a bug on retrieving the status fro UI. And your version of airflow is pretty old. I would recommend you to upgrade to latest version, try there, and if you see it still a problem, report an issue. I think the problem might be in just visualising the dependency in this case and this part may have been fixed already.