[Python] [Airflow]: Dynamic Task Mapping on DockerOperator using Xcoms

I am creating a dag that should do the following:

• fetch event ids
• for each event id, fetch event details ( DockerOperator )

The code below is my attempt to do what I want:

from datetime import datetime

from airflow.operators.python import PythonOperator
from airflow.providers.docker.operators.docker import DockerOperator



With Dag(
start_date=datetime(2024, 11, 1),
schedule="@daily",
):
task_fetch_ids = PythonOperator(
task_id="fetch_detail",
...)


task_fetch_detail = DockerOperator(
task_id="fetch_detail",
image="image:v1",
).expand(
command=[f"fetch-event --event-id {event_id}" for event_id in "{{ ti.xcom_pull(task_ids='task_fetch_ids', key='return_value') }}"]
)


task_fetch_ids >> task_fetch_detail




The above clearly doesn't work because I am looping through a string. What is the correct syntax ?

Continue reading...