Automating CSV Imports Into a Database Without Surprises

The first version of a CSV import is a five-line script that someone runs by hand on a Monday. The second version is the same script in cron, and it works for three weeks. The third version is the one that gets written after a Tuesday morning when the provider's file arrived half-written, the job loaded it, a dashboard showed revenue down by 40 percent and nobody found out until a customer asked. Everything in this guide is the difference between the second version and a job you can leave alone. It covers the shape of an unattended importer, a complete working example, the decisions that need to be made before the first bad file, and where each part is covered in more detail. If you only want the load command for one database, start with loading a CSV file into PostgreSQL or importing a CSV into MySQL; this is about running that command every day without looking at it.

The shape of an unattended import

Every reliable importer I have read does the same six things in the same order, whatever language or database it uses. The names differ, and the order does not.

The stages are separate on purpose. Detection looks at the file system. Validation looks at the file's bytes. Loading looks at the database. Treating them as one block of code is how an error in the last step leaves the first three undone or half-done. Most of the real bugs I have seen come from a stage that was left out and not from one written badly: no detect, so a truncated upload loaded; no record, so a rerun duplicated. This guide walks through the stages with one concrete script, then takes each decision in turn.

A complete importer, in one file

The script below loads order files named orders-YYYY-MM-DD.csv from an incoming folder into a SQLite database, once each. SQLite stands in for any transactional database so that anyone can run it, and the comments mark where a PostgreSQL or MySQL version differs. It is about eighty lines and I would rather you read it in full than a summary of it.

#!/usr/bin/env python3
"""Load order files from incoming/ into SQLite, once each, and keep a ledger."""
import csv, hashlib, shutil, sqlite3, sys
from datetime import date, datetime, timezone
from pathlib import Path

EXPECTED = ["order_id", "customer", "placed_on", "total"]
INCOMING, ARCHIVE, REJECTED = Path("incoming"), Path("archive"), Path("rejected")

SCHEMA = """
CREATE TABLE IF NOT EXISTS orders (
    order_id  INTEGER PRIMARY KEY,
    customer  TEXT NOT NULL,
    placed_on TEXT NOT NULL,
    total     REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS import_log (
    sha256    TEXT PRIMARY KEY,
    file_name TEXT NOT NULL,
    rows_ok   INTEGER NOT NULL,
    rows_bad  INTEGER NOT NULL,
    loaded_at TEXT NOT NULL
);
"""

def sha256_of(path):
    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 parse(row):
    order_id = int(row["order_id"])
    placed_on = date.fromisoformat(row["placed_on"]).isoformat()
    total = float(row["total"])
    if not row["customer"].strip():
        raise ValueError("empty customer")
    if total < 0:
        raise ValueError("negative total")
    return order_id, row["customer"].strip(), placed_on, total

def load(db, path, digest):
    good, bad = [], []
    with open(path, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        if reader.fieldnames != EXPECTED:
            raise SystemExit(f"{path.name}: header {reader.fieldnames} is not {EXPECTED}")
        for row in reader:
            try:
                good.append(parse(row))
            except (ValueError, TypeError) as e:
                bad.append((reader.line_num, str(e), row))
    with db:  # one transaction: all rows and the ledger entry, or nothing
        db.executemany(
            """INSERT INTO orders VALUES (?,?,?,?)
               ON CONFLICT(order_id) DO UPDATE SET
                 customer=excluded.customer, placed_on=excluded.placed_on, total=excluded.total""",
            good,
        )
        db.execute(
            "INSERT INTO import_log VALUES (?,?,?,?,?)",
            (digest, path.name, len(good), len(bad), datetime.now(timezone.utc).isoformat(timespec="seconds")),
        )
    if bad:
        REJECTED.mkdir(exist_ok=True)
        with open(REJECTED / (path.name + ".rejects.csv"), "w", newline="") as out:
            w = csv.writer(out)
            w.writerow(["line", "reason", *EXPECTED])
            for line, why, row in bad:
                w.writerow([line, why, *(row[k] for k in EXPECTED)])
    return len(good), len(bad)

def main():
    db = sqlite3.connect("shop.db")
    db.executescript(SCHEMA)
    ARCHIVE.mkdir(exist_ok=True)
    files = sorted(INCOMING.glob("orders-*.csv"))
    if not files:
        print("no new files")
    for path in files:
        digest = sha256_of(path)
        seen = db.execute("SELECT file_name FROM import_log WHERE sha256 = ?", (digest,)).fetchone()
        if seen:
            print(f"{path.name}: skipped, same content already loaded as {seen[0]}")
            shutil.move(path, ARCHIVE / ("duplicate-" + path.name))
            continue
        ok, bad = load(db, path, digest)
        shutil.move(path, ARCHIVE / path.name)
        print(f"{path.name}: loaded {ok} rows, rejected {bad}")

if __name__ == "__main__":
    sys.exit(main())

Two sample files go into incoming. The first is clean. The second corrects the total of order 1003, adds a good row, and carries two bad ones: an empty customer and a date in month 13.

order_id,customer,placed_on,total
1003,Cy Foods,2026-10-01,17.50
1004,Delta Inc,2026-10-02,200.00
1005,,2026-10-02,10.00
1006,Echo LLC,2026-13-02,45.00

Run it, run it again, then copy the second file back into incoming under a new name and run it a third time:

orders-2026-10-01.csv: loaded 3 rows, rejected 0
orders-2026-10-02.csv: loaded 2 rows, rejected 2
---
no new files
---
orders-2026-10-03.csv: skipped, same content already loaded as orders-2026-10-02.csv

The first run wrote five rows from two files and left four in the table, because the second file's correction of order 1003 (total from 15.00 to 17.50) updated the existing row instead of adding a second one. The second run had nothing to do. The third run recognized the renamed file by its content and refused to load it again. The rejected rows landed in a file of their own with the reason and the line number:

line,reason,order_id,customer,placed_on,total
4,empty customer,1005,,2026-10-02,10.00
5,"month must be in 1..12, not 13",1006,Echo LLC,2026-13-02,45.00

That is a working import job. Everything else in this guide is about what the script does not yet handle, and about the decisions it makes quietly.

Why the pieces are shaped this way

The ledger is keyed by content, not by name. The primary key of import_log is the SHA-256 of the file. A provider who re-sends Tuesday's file under a new name, or a person who copies a file into the folder twice, cannot cause a second load. Keying on the file name fails the first time a file is corrected and re-sent under the same name, and keying on a date fails when two files cover the same day. The hash answers exactly one question, "have I already loaded these bytes?", and it answers it right. Idempotent CSV imports that are safe to run twice goes through the other designs and why each one breaks.

The load and the ledger row share one transaction. The with db: block commits the upserted rows and the ledger entry together. If the process dies between them, neither exists, and the next run starts over on the same file. Without that, a crash after the load and before the log leaves rows in the table and no record, so the next run loads them again; a crash in the other order records a load that never happened. Atomicity is the property that makes "run it again" a valid answer to every failure.

The merge is an upsert on a business key. ON CONFLICT(order_id) DO UPDATE means a corrected order replaces the old one. PostgreSQL has the same INSERT ... ON CONFLICT, and MySQL has INSERT ... ON DUPLICATE KEY UPDATE. For large files, load into a staging table first and merge from it in one statement (loading a CSV into a staging table, then merging), because a million single-row upserts is far slower than one set-based merge.

Bad rows go to a file. The job did not stop at the first bad row, and it did not discard the bad rows silently. It wrote them with line numbers to a file someone can read. Logging rejected rows from a CSV import for review covers the design, including how to avoid logging personal data in the reject file.

The archive is a move, after the commit. A file leaves incoming only once its load is committed. A crash before the move means the file is picked up again, and the ledger recognizes it. The archive keeps the original bytes, which is what you need when someone asks next month why a number changed (archiving processed CSV files automatically).

Knowing a file is complete

The script reads whatever is in incoming, and that is its most dangerous assumption. A provider that uploads by SFTP writes a file over several seconds or minutes, and a job that starts mid-upload reads a truncated file that is perfectly valid CSV up to the cut. It will load, it will pass validation, and the missing rows will not be noticed until someone compares totals.

The reliable protocols are three, in order of how much I trust them. An atomic rename: the producer writes orders.csv.part and renames it to orders.csv when done, and your glob matches only the final name. A rename on one file system is atomic, so a visible file is a whole file. A done marker: the producer writes a small orders-2026-10-02.done file (or a manifest with the row count and a checksum) after the data file, and the job processes a data file only when its marker exists. The manifest also gives you a check of your own: the number of rows you loaded must equal the number it names. A stability check: if neither is possible, ignore files modified in the last few minutes and require the size to be the same across two looks. It is the weakest of the three, because a slow upload can pause for longer than your window.

Ask the provider for the first or second. If you cannot, the third plus a row-count check against the previous days is the fallback, and you should say so in the runbook. Watching a folder for new CSV files and importing them covers the event-based version, and testing that a CSV export was not truncated lists checks that catch the cut-off file after the fact.

Reject the file or reject the row

The example loads the good rows of a partly bad file. That is a policy, and it is not always the right one. Decide it before the first bad file, and write it down.

Row-level rejection suits event-like data where each row is independent: page views, sensor readings, log lines. A few bad rows among a million are noise, the good rows are valuable, and a reject file carries the rest. File-level rejection suits data that only makes sense as a whole: a daily balance, a full price list, a snapshot that replaces the table. One missing row in a snapshot means the next merge deletes that record, and a half-loaded snapshot is worse than a late one.

A middle path is a threshold: load if fewer than 0.1 percent of rows are bad, and reject the file otherwise, on the theory that a flood of bad rows means the producer changed something. Whatever you pick, the job must tell a human, because a file that was partly rejected is a data-quality event. The rate of rejections over time is also a useful number: a steady two a day is a source with a known quirk, and 40 percent is a changed format.

Choosing a scheduler

The scheduler only decides when the script starts. The choice matters less than what the script does when started twice, or not at all.

For most single-server setups, cron plus the importer above is correct, and the rest of the guide is what makes it dependable. Scheduling CSV reports from SQL with cron covers the cron side, including the environment trap that makes a job work in your shell and fail at 02:00.

Overlap, locks and retries

If a run takes longer than the interval between runs, a second copy starts while the first is still working. With the ledger and the transaction, two copies cannot corrupt the data, but they can both load the same file, and the second one fails with a primary key error on the ledger. That is acceptable. A cleaner approach is to prevent overlap: flock -n /var/lock/import_orders.lock python3 import_orders.py in cron runs the job only if it can take the lock, and exits at once otherwise. In PostgreSQL, pg_try_advisory_lock does the same inside the database, which also covers jobs on different machines.

Retries need care. A job that retries on any exception will hammer a database that is down, and will retry a bad file forever. Retry only the errors that are likely to be temporary (a connection refused, a lock timeout), a few times with growing delays, and never retry a validation failure. A file that fails validation moves to a failed folder with the reason, and waits for a person.

Schema drift and providers who change things

Providers change columns without telling you. The sample job refuses a file whose header differs from the expected list, which is the safe default: it stops and says what it saw. The alternative of reading columns by position loads wrong data quietly, and the alternative of reading by name and ignoring extras loads fine until the extra is a renamed column you needed. Write the expected header in code, compare exactly, and fail loudly. Handling a CSV provider who changes columns without warning covers the cases where you need to adapt instead of fail, and making a CSV a contract with the data provider covers getting the provider to tell you first.

Types drift too. A column that held integers for a year begins to carry N/A. Row-level validation catches it; so does a typed staging table whose load fails on the first bad value. Pair the two: a lenient parse that records the rejects, and a strict final table that refuses what slipped through.

Seeing what happened

A job that works silently also fails silently. Three things make it observable.

A ledger you can query. The import_log table answers "what loaded, when, and how many rows" with one query. Add the elapsed time and the source path if you have more than one feed. A freshness check. An alert that fires when no file has loaded for 26 hours catches the failure that no exception reports, such as the provider not sending, the cron daemon not running, or the file landing in the wrong folder. It is the single most valuable check, and it is one query:

SELECT now() - max(loaded_at) AS age FROM import_log;

A row-count comparison. Compare today's rows with the average of the last seven. A file with 8 percent of the usual rows is probably truncated, whatever its checksum says. The data-quality checks in writing data quality checks for a CSV fit here, run after the load.

Send alerts to a place a person looks, with the file name and the reason in the message, and send a success summary once a day so that silence means something.

Permissions and secrets

The importer needs the least privilege that works: insert and update on its target tables, insert on the ledger, and nothing else. It does not need superuser, file access on the database server or the right to create tables. Keep the connection string in an environment variable or a secrets manager and never in the script or the repository. On the file side, the account that runs the job reads incoming and writes archive and rejected, and the provider's account can write incoming and read nothing. The wider case for these boundaries is in CSV security risks that catch teams out, and the upload limits that apply to a feed from outside are in limiting file size and row count for CSV uploads.

Testing the job

An importer is easy to test, and nearly nobody does. Keep a folder of fixture files: a clean one, one with each kind of bad row, one with a changed header, an empty file, a file with only a header, a file with a byte order mark and a truncated file. A test runs the importer against an empty database and asserts the row counts and the ledger. The most valuable test is the replay: run the same fixtures twice and assert that the second run changes nothing. A test for the corrected file, same business keys with new values, shows the upsert does what you think.

Plan the recovery path while calm. A backfill is the importer pointed at a folder of old files, and with a ledger it is safe: files already loaded are skipped, missing days load, and order does not matter as long as each file is a delta. If your files are snapshots and order does matter, add the file's date to the ledger and refuse to load an older snapshot over a newer one.

Failure modes, and what the job does about each

It is worth walking the list of things that go wrong, because each one maps to a stage above, and a job that handles them by design needs no heroics at 02:00.

The file is empty, zero bytes. Validation fails on the missing header, the job moves it to a failed folder and alerts. It does not load nothing and call it a success, because an empty file from a provider that normally sends 50,000 rows is a failure on their side that you want to hear about the same day.

The file has only a header. This is subtler, since it is valid CSV and a legitimate "no rows today" for some feeds. Your job needs a rule per feed: allowed or not. If it is allowed, record it as a load of zero rows so the freshness check passes; if not, treat it like an empty file.

The file is the same as yesterday's. A provider job that reruns its previous export is common, and the content hash catches it. The job logs a skip and the alert says "unchanged file", because a feed that sends the same bytes two days running is stuck.

The header changed. The job refuses and names both headers. Someone decides whether to adapt the code, ask the provider to revert, or map the new name to the old one, and that decision is a code change with a review, not a runtime guess.

One row is malformed, with a stray quote that swallows the next hundred lines into one field. The parser either raises an error with a line number or, in lenient mode, yields a row with the wrong field count. The field-count check catches the second case and the field-size limit catches the first. The job treats a quoting error as file-level damage, since the rows after it cannot be trusted.

The database is down or the connection drops mid-load. The transaction rolls back, the ledger has no row, the file stays in incoming, and the next run tries again. This is the case the single transaction was built for.

The disk is full. The archive move fails after the commit. The rows are loaded, the ledger says so, and the file is still in incoming, so the next run sees a known hash, logs a skip and tries the move again. It is also the reason to alert on free space, since a full disk stops everything at once.

A corrected file arrives under the same name. Its hash is new, so it loads; the upsert updates changed rows; the ledger shows two files with the same name and different hashes, which is exactly the history you want.

The job runs twice at the same time. With the lock, the second exits. Without it, one wins and the other fails on the ledger's primary key. Neither corrupts anything.

What is left, the failures that need a human, is short: the provider's data is wrong but valid, a new column carries meaning, the schema needs a migration. Those are the jobs of the people who own the feed, and the import job's role is to make them visible quickly.

Encodings, delimiters and dates at the door

A fixed set of format surprises hits every feed, and the cheapest place to handle them is one function at the start of the pipeline. Open files with utf-8-sig, as the sample does, so a byte order mark is dropped and does not become part of the first column name (the UTF-8 BOM in CSV files explains why that matters). If the provider sends Windows-1252, state it in the feed's configuration and decode explicitly, rather than detect on every file and risk guessing wrong (detecting the encoding of a CSV file). Fix the delimiter in configuration too; a sniffer is a good tool for exploring a file and a poor one for a production feed, since a file with few rows can fool it.

Dates deserve the strictest treatment. Parse them with an explicit format and reject anything else, because the lenient parser that accepts both 03/04/2026 and 2026-04-03 is the one that swaps day and month once a year. The sample job calls date.fromisoformat, which accepts only one shape. Numbers get the same rule: if the provider uses a decimal comma, convert it explicitly, and reject a thousands separator you did not expect. Strictness at the door is what lets everything after it be simple.

A runbook of one page

Automation moves the work from doing to responding, and a response needs a page of instructions. For each feed, write down: where the files come from and who to contact, what the schedule is, where the job runs and how to run it by hand, where the logs and the ledger are, what each alert means and what to do about it, how to reload a day, and how to disable the job. Put it in the repository next to the code, so it is updated by the people who change the code.

The test of a runbook is whether someone who did not write the job can use it. Hand it to a colleague and watch them reload yesterday's file. Every place they get stuck is a missing sentence. The hour this takes is cheaper than the first incident without it, and it makes the feed something the team owns and not something one person knows.

What to graph

If you have a dashboard, five series give you almost everything: files loaded per day, rows loaded per file, rows rejected per file, seconds from file arrival to load completion and the age of the newest ledger row. Plot them for a month and the feed's rhythm becomes visible: the weekday pattern, the typical size and the normal reject rate. After that, anomalies are easy to define as a distance from the pattern and not as a constant someone guessed.

None of these needs a monitoring product to start. A daily summary email built from the ledger, listing yesterday's files, row counts and rejects, is read by people and costs one query. The graphs can wait until the summary has proven useful.

Speed, when it matters

The sample job inserts in a single executemany, which is fine for tens of thousands of rows a day. At millions, switch to the database's bulk path: COPY in PostgreSQL (the COPY command for CSV), LOAD DATA in MySQL (LOAD DATA INFILE), .import in SQLite (the SQLite CSV import). The structure stays the same: bulk load into staging, merge in one statement, write the ledger in the same transaction. For files that live in object storage, loading a CSV from S3 into a database shows how to keep the same shape without copying files to a server first, and a Python script that loads a daily CSV into PostgreSQL is the PostgreSQL version of this job with the driver's copy stream.

When a script is the wrong tool

A hand-written importer stops being the right tool at a few edges. If you have dozens of feeds with different formats and owners, the maintenance cost of dozens of scripts exceeds the cost of a managed pipeline tool, and the tool's connectors, retries and lineage start to pay. If data must reach the database within seconds of an event, a file-based job is the wrong shape and a stream or a change-data-capture feed is the right one. If the volume is billions of rows, the loader should live inside the warehouse and read files from object storage directly, and the warehouse's own ingestion features (loading CSV into cloud warehouses) do the job better than any client script.

And if the feed is one file a month that a person looks at anyway, a documented manual procedure beats an automation nobody will remember how to fix. Automate the feeds that run often enough that a human forgetting is the dominant risk. For everything else, write the script as if you will hand it to a stranger, put the six stages in it, and keep the ledger. The format itself leaves everything above to you; RFC 4180 describes the file and not the process, and the database manuals for PostgreSQL's COPY and SQLite's upsert are the references for the statements used here.