Loading CSV Files in Airflow With a Plain Python Task

Airflow has an operator for most things, and for loading a CSV into a database the natural-looking chain is a file sensor, a transfer operator and a SQL operator. It works, and it also scatters one small job across four configuration objects, none of which you can run on your laptop without a scheduler. For a daily feed I would rather write one Python function that does the load, test it like any other function, and have Airflow do what it is good at: run it on a schedule, retry it on failure and show whether it ran. This article shows that shape. It assumes the importer design from automating CSV imports into a database and the PostgreSQL loader from a Python script that loads a daily CSV into PostgreSQL.

The function first

Start outside Airflow. The load is a function that takes a file path and returns the number of rows merged, with no dependence on any Airflow object. It is the same function you would run from cron:

from pathlib import Path

def load_orders(path: Path, conn) -> int | None:
    """Stage, merge and record one file. Returns rows merged, or None if already loaded."""
    ...  # see the PostgreSQL loader: hash check, COPY into stage, upsert, ledger row

Keep it in a module the DAG imports. That module can have unit tests with a fixture file and a scratch database, and a person can run it by hand during an incident without finding out how to trigger a task. This is the single most useful decision in the design: the task is a thin wrapper, and the logic is somewhere you can test it.

The DAG

The DAG finds the day's file, loads it and checks the result. Three tasks is plenty:

from datetime import datetime
from pathlib import Path

from airflow.sdk import dag, task
from airflow.providers.postgres.hooks.postgres import PostgresHook

from orders_loader import load_orders

INCOMING = Path("/data/incoming")

@dag(
    schedule="0 6 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args={"retries": 2},
    tags=["csv", "orders"],
)
def load_daily_orders():
    @task
    def find_file(**context) -> str:
        day = context["logical_date"].date().isoformat()
        path = INCOMING / f"orders-{day}.csv"
        if not path.exists():
            raise FileNotFoundError(f"{path} has not arrived")
        return str(path)

    @task
    def load(path: str) -> int:
        hook = PostgresHook(postgres_conn_id="shop_db")
        with hook.get_conn() as conn:
            merged = load_orders(Path(path), conn)
        return 0 if merged is None else merged

    @task
    def check(merged: int) -> None:
        if merged == 0:
            print("file was already loaded, nothing to check")

    check(load(find_file()))

load_daily_orders()

Read it top to bottom. find_file derives the file name from the run's logical date, which is the date the run represents, not the moment it happens to execute. A rerun of 1 October tomorrow therefore looks for the 1 October file, which is what a backfill needs. If the file is not there, the task fails and Airflow retries it, which handles a provider that is a few minutes late. load takes the path, opens a connection from an Airflow Connection named shop_db (credentials live in Airflow's secrets backend, not in the code) and calls the plain function. The return values are a path and an integer, both tiny.

Keep data out of XCom

Airflow passes task return values through XCom, which stores them in the metadata database. A path is fine. A million rows are not: they bloat the database, slow the scheduler and fail in confusing ways. The rule is to pass pointers and counts, and keep the data in the file system, the object store or the target database. If a later task needs the rows, it reads them from where the earlier task put them. The same rule applies if you use operators: do not return a result set from a task.

Make retries safe

Retries are the reason to use a scheduler, and they are only safe if the task can run twice. The loader's content-hash ledger gives that: a retry after a task died mid-load either finds nothing committed and loads, or finds the ledger row and returns None. Idempotent CSV imports that are safe to run twice explains why the ledger and the load must commit together.

Decide what is retried. A connection refused, a lock timeout and a file that has not arrived yet are temporary and worth two or three retries with a delay. A header that does not match, a value that fails a constraint, a file that is empty: these will fail the same way every time, and retrying them only delays the alert. Raise a distinct exception for validation failures and use AirflowFailException so Airflow fails the task without retrying:

from airflow.exceptions import AirflowFailException

raise AirflowFailException(f"{path.name}: header is not the expected one")

AirflowFailException marks the task failed at once, which is the right outcome for a file a person must look at.

Catchup and backfill

catchup=False means that if the scheduler was down for three days, it runs only the most recent interval when it returns. For a feed where yesterday's file matters only until today's arrives, that is right. For a feed where each day must be loaded, set catchup=True and let Airflow create a run for each missed date, which works because each run's file name comes from its logical date and each load is idempotent. To replay a range after a bug fix, clear the task instances for those dates in the UI, or run airflow backfill create on Airflow 3; either way the ledger skips days that already loaded and fills the rest.

Concurrency needs a thought if a backfill creates thirty runs at once. Set max_active_runs=1 on the DAG so a day is loaded before the next starts, which also protects a target table from parallel upserts of overlapping keys.

Test it without a scheduler

Because the logic is a function, the tests are ordinary:

• Call load_orders on a fixture file twice and assert that the second call returns None and the table is unchanged.

• Call it on a file with a wrong header and assert the failure.

• Import the DAG module and assert that it parses: Airflow's DagBag or a simple import in a test catches syntax errors and missing imports before deploy.

• Run airflow dags test load_daily_orders 2026-10-02 to execute a whole run locally with the scheduler out of the loop.

A DAG file that imports heavy modules at the top slows the scheduler, which parses every DAG file continuously. Keep the top level light and import expensive things inside the task function.

When Airflow is more than the job needs

Airflow is a service with a database, a scheduler and a web server. For one daily CSV load, that is a lot of infrastructure to maintain, and cron with the importer script is smaller and has fewer ways to fail (scheduling CSV reports from SQL with cron). Airflow starts to pay when you have many jobs with dependencies ("load orders, then build the report, then email it"), when you need a UI that shows what ran and why it failed, or when a team needs retries, alerts and backfills as shared features and not as per-script code.

If you already run Airflow, use it; the plain-function pattern keeps the logic portable, so moving the job out later costs a wrapper. If you do not, do not start with Airflow for this. The same advice applies to the heavier alternative of declaring the load in a transformation tool, where seeding a database from CSV files with dbt seeds fits reference data and not daily feeds. The authority on the decorators and context used above is the Airflow documentation, and the PostgreSQL hook is described in the Postgres provider documentation.