A Python Script That Loads a Daily CSV Into PostgreSQL

The daily file shows up at six in the morning, someone has to get it into PostgreSQL, and that someone should not be you. The script below is the one I would write for that: about sixty lines, one dependency, no framework. It checks the header, loads the file through COPY into a temporary table, merges into the real table with an upsert and writes a ledger row, all in one transaction, so a failure at any step leaves the database as it was and a rerun is harmless. It is the PostgreSQL version of the importer in automating CSV imports into a database, and it assumes you have read the idea there and want the code here.

The tables

The script needs two tables, a target and a ledger. Create them once:

CREATE TABLE orders (
    order_id  integer PRIMARY KEY,
    customer  text NOT NULL,
    placed_on date NOT NULL,
    total     numeric(10,2) NOT NULL
);

CREATE TABLE import_log (
    sha256      text PRIMARY KEY,
    file_name   text NOT NULL,
    rows_merged integer NOT NULL,
    loaded_at   timestamptz NOT NULL DEFAULT now()
);

The primary key on orders is the business key the upsert uses. The primary key on import_log is the content hash of the file, which is what makes a rerun a no-op and not a duplicate. If your data has no natural key, idempotent CSV imports that are safe to run twice shows other ways to get one.

The script

Here is the whole thing. The load is a function that takes an open connection and a path, so it is also easy to call from a test.

#!/usr/bin/env python3
"""Load one daily orders CSV into PostgreSQL: stage with COPY, merge, record."""
import csv, hashlib, os, sys
from pathlib import Path

import psycopg

EXPECTED = ["order_id", "customer", "placed_on", "total"]

def sha256_of(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()

def check_header(path: Path) -> None:
    with open(path, newline="", encoding="utf-8-sig") as f:
        header = next(csv.reader(f), [])
    if header != EXPECTED:
        raise SystemExit(f"{path.name}: header {header} is not {EXPECTED}")

def load(conn: psycopg.Connection, path: Path) -> int | None:
    digest = sha256_of(path)
    with conn.transaction(), conn.cursor() as cur:
        cur.execute("SELECT file_name FROM import_log WHERE sha256 = %s", (digest,))
        if cur.fetchone():
            return None                      # same bytes already loaded
        cur.execute(
            "CREATE TEMP TABLE stage (LIKE orders INCLUDING DEFAULTS, line bigserial) ON COMMIT DROP"
        )
        copy_sql = (
            "COPY stage (order_id, customer, placed_on, total) "
            "FROM STDIN WITH (FORMAT csv, HEADER true)"
        )
        with open(path, "rb") as f, cur.copy(copy_sql) as copy:
            while chunk := f.read(1 << 16):
                copy.write(chunk)
        cur.execute(
            """
            INSERT INTO orders (order_id, customer, placed_on, total)
            SELECT DISTINCT ON (order_id) order_id, customer, placed_on, total
            FROM stage
            ORDER BY order_id, line DESC
            ON CONFLICT (order_id) DO UPDATE
              SET customer = EXCLUDED.customer,
                  placed_on = EXCLUDED.placed_on,
                  total = EXCLUDED.total
            """
        )
        merged = cur.rowcount
        cur.execute(
            "INSERT INTO import_log (sha256, file_name, rows_merged) VALUES (%s, %s, %s)",
            (digest, path.name, merged),
        )
        return merged

if __name__ == "__main__":
    path = Path(sys.argv[1])
    check_header(path)
    with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
        merged = load(conn, path)
    print(f"{path.name}: already loaded" if merged is None else f"{path.name}: merged {merged} rows")

Here is a file of four data lines, with order 1003 sent twice, the second time with a corrected total:

order_id,customer,placed_on,total
1001,Acme Ltd,2026-10-01,120.50
1002,Beta Corp,2026-10-01,89.90
1003,Cy Foods,2026-10-01,15.00
1003,Cy Foods,2026-10-01,17.50

Run it twice, then look at the tables:

export DATABASE_URL=postgresql://[email protected]/shop
python3 load_daily.py orders-2026-10-02.csv
python3 load_daily.py orders-2026-10-02.csv
psql -c "select * from orders order by 1" -c "select file_name, rows_merged from import_log"

orders-2026-10-02.csv: merged 3 rows
orders-2026-10-02.csv: already loaded
 order_id | customer  | placed_on  | total  
----------+-----------+------------+--------
     1001 | Acme Ltd  | 2026-10-01 | 120.50
     1002 | Beta Corp | 2026-10-01 |  89.90
     1003 | Cy Foods  | 2026-10-01 |  17.50
(3 rows)

       file_name       | rows_merged 
-----------------------+-------------
 orders-2026-10-02.csv |           3
(1 row)

Four data lines became three rows, because the duplicate key 1003 was collapsed to its last occurrence (17.50) before the merge, and the second run found the file's hash in the ledger and did nothing. The password does not belong in the URL; use a .pgpass file or the PGPASSWORD environment variable set by whatever launches the job, so it never appears in a process list.

The parts worth explaining

**cur.copy() is the loader.** psycopg 3 exposes the COPY protocol directly. You open a copy block with the COPY ... FROM STDIN statement and write bytes to it; the driver streams them to the server. Reading the file in 64 KB chunks keeps memory flat on a 5 GB file, and a row-by-row INSERT loop would be one to two orders of magnitude slower. The format, quoting and header handling are the server's, which is the point: PostgreSQL's CSV parser follows the rules described in the COPY command for CSV and you do not have to re-implement them in Python.

The header is checked in Python. HEADER true tells COPY to discard the first line, and it does not look at it. A file with the columns in a different order loads without an error and puts the customer's name in the ID column, or fails with a type error that points at a row and not at the cause. The check_header function compares the names to the expected list and stops with a message that shows both. It opens the file with utf-8-sig, so a byte order mark does not turn order_id into order_id.

The stage table is temporary and drops on commit. LIKE orders INCLUDING DEFAULTS copies the column definitions, and not the primary key, so the stage table accepts duplicate keys. The extra line bigserial column records the order the rows arrived in. ON COMMIT DROP removes the table at the end of the transaction, whether it commits or rolls back, so a crashed run leaves no stage tables behind.

**DISTINCT ON keeps the last row for each key.** If a file contains order 1003 twice, a plain INSERT ... SELECT ... ON CONFLICT DO UPDATE raises ON CONFLICT DO UPDATE command cannot affect row a second time, because one statement may not update the same target row twice. Selecting DISTINCT ON (order_id) ordered by line DESC hands the merge one row per key, the one that appeared last in the file, which matches how most people read "later rows win". If your rule is different, such as "reject the file on duplicate keys", test for duplicates first and fail.

The ledger row is in the same transaction. A crash between the merge and the ledger insert rolls both back. That property is the whole point, and it is why the transaction wraps everything from the ledger check to the ledger write. A common mistake is to commit the load and write the ledger in a second step, which leaves a window where the file is loaded and unrecorded.

What it does not do

It does not validate values beyond what the column types enforce. A date column rejects 2026-13-02 and a numeric column rejects eighty, and either error aborts the whole transaction with a message naming the row. That is right for a file you trust and wrong for one where a few bad rows are expected. For the second case, load the stage table with text columns, validate in SQL, and copy rejects to a table (logging rejected rows from a CSV import for review). It does not move the file afterward, which is a separate step (archiving processed CSV files automatically). And it does not retry, since a retry policy belongs to whatever schedules it.

Permissions are narrow by design. The role in DATABASE_URL needs INSERT and UPDATE on orders, INSERT and SELECT on import_log and the right to create temporary tables, and nothing on the server's file system, because COPY FROM STDIN reads from the client. Do not give the role pg_read_server_files to make an import work, which is the habit CSV security risks that catch teams out warns about.

When the file is too clean or too dirty for this script

If the files are always clean and you only need them loaded once, the script is more than you need: psql -c "\copy orders FROM 'orders.csv' CSV HEADER" is the whole job (loading a CSV file into PostgreSQL). If the files are always dirty, with quoting errors, shifting columns and mixed encodings, this script will stop on the first one, and the right shape is a lenient first pass that records what it could not read. A file in a different encoding needs ENCODING 'WIN1252' in the COPY options or a conversion step (converting a CSV from Latin 1 or Windows 1252 to UTF8).

For schedulers, the script exits non-zero on failure and prints one line on success, which is what cron, systemd and Airflow all want. Scheduling CSV reports from SQL with cron covers the environment details that make a cron job behave like your shell. The references for the statements above are the PostgreSQL manual pages for COPY and INSERT with ON CONFLICT, and the psycopg documentation on using COPY.