[Python] Airflow DAG with multiple dynamic SQLExecuteQueryOperator tasks keeps breaking...

I'm trying to move a CSV file into a MariaDB database. To do this, I wrote the following DAG.

When I run this through the Airflow UI, it works fine.

But then the DAG disappear from the list, and I don't know why. Is there something obvious I'm missing here ?

from datetime import datetime

from airflow import DAG
from airflow.decorators import dag,task
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.operators.dummy import DummyOperator

import pandas as pd

default_args = {
"email_on_failure": False,
"email_on_retry": False,
"retries": 0,
"start_date": datetime(2025,1,1),
"catchup": False,
"schedule": 'None',
}

@dag(
default_args=default_args,
dag_display_name="Load all Donations"
)
def opportunity_load():
all_tasks = []
chunk_size = 100

# We load the contact CSV in chunks
cont_counter = 0

for cont_chunk in pd.read_csv('/opt/airflow/data/Contacts.csv',chunksize=chunk_size):
query_parameters = {"affiliate":"uk","contacts":[]}
for index,row in cont_chunk.iterrows():
query_parameters["contacts"].append({
"salesforce_id" : row['Id'],
"first_name" : row['FirstName'],
"last_name" : row['LastName']
})
cont_counter += 1

current_task = SQLExecuteQueryOperator(
task_id = f'cont_push_chunk_{cont_counter}',
sql = {'load_contacts.sql'},
conn_id = 'datawarehouse',
params = query_parameters
)
all_tasks.append(current_task)

for task_i in range(len(all_tasks) - 1):
all_tasks[task_i] >> all_tasks[task_i + 1]

opportunity_load()


I replaced the SQLExecuteQueryOperator with dummy tasks, and that fixed the issue, so I'm assuming there's an issue with my SQL task somewhere. The strange thing is that the DAG still works in the command line.

As you can see I break down the CSV into chunks of 100 (I'll probably increase this to 1000+ later). Then I push each chunk to my database using a query pattern stored in a file called load_contacts.sql, which you can see below:

INSERT INTO `{{params.affiliate}}_contacts`(salesforce_id,first_name,last_name)
VALUES
{% for contact in params.contacts %}
("{{contact.salesforce_id}}","{{contact.first_name}}","{{contact.last_name}}")
{% if not loop.last %}
,
{% endif %}
{% endfor %}
ON DUPLICATE KEY UPDATE
salesforce_id=VALUES(salesforce_id),
first_name = VALUES(first_name),
last_name = VALUES(last_name);


This is all on a local machine using Docker.

Continue reading...