Python ETL pipeline: load a daily CSV into PostgreSQL

A daily CSV import is a useful first Python ETL pipeline: read a file, validate and transform its records, then load them into a PostgreSQL table. The difficult parts are usually the boundaries. What happens when a date is invalid, an order appears twice, the database rejects a write, or yesterday’s job runs again?

This walkthrough builds a small orders import with explicit types, a rejection report, and a transactional upsert. Start with the synthetic file provided here and a disposable database. Use the existing PostgreSQL Docker Compose guide for the private database and application role; the optional adapter below connects this job to that lab.

Test limits, November 11, 2026: 18 local Python tests and a no-database validation run passed on Windows. The merged Compose configuration was checked, and Psycopg 3.3.8 was installed and imported. PostgreSQL connections, SQL, commit/rollback behavior, database reruns, lock contention, container execution, scheduling, recovery, and cloud deployment were not tested. Database outputs below are expectations to verify, not captured results.

Define the ETL workflow and its input contract

ETL means extract, transform, and load. Here, extraction reads a completed UTF-8 CSV file; transformation validates IDs, dates, and USD amounts; loading writes accepted records to PostgreSQL. Unlike ELT, the validation and type conversion happen before the source data enters the destination table. This example is a batch job, not a streaming system or a data warehouse platform.

The input must contain exactly these columns in this order. Save the supplied fixture as orders.csv:

order_id,order_date,amount_usd
ORD-1001,2026-11-01,19.95
ORD-1002,2026-11-01,80.00
ORD-1003,2026-11-02,0.05

An order ID is the stable business key. Dates use YYYY-MM-DD, and nonnegative USD amounts have exactly two decimal places, up to 99,999,999.99. The program trims field whitespace, converts dates to Python date objects, and uses Decimal for amounts. It does not infer currencies, replace missing values, or silently round malformed amounts.

Duplicate IDs within one file are rejected, even if the two records look identical. If any record is rejected, the entire file is held back from loading. Fix or reconcile the source before trying again. This policy avoids publishing a partly imported daily report without making that choice explicit.

The file is limited to 10 MiB and 10,000 records. These are teaching limits, not measured capacity. The sample keeps the bounded input and validated rows in memory. For larger data volumes, design chunking, staging, and transaction boundaries separately; adding more CPU does not remove a memory or storage constraint.

Download the sample and run validation first

The ETL sample archive contains the complete application, schema, synthetic CSV, 18-test suite, and optional Docker/Compose files. The code excerpts in this article come from those files; use the complete archive for the commands. Keep a separate copy of the original input for reconciliation.

With Python 3.12 available, open a terminal in the extracted sample directory. The validation path and tests use the standard library and do not require a database or the PostgreSQL client package:

python etl.py orders.csv
python -m unittest -v test_etl.py

The observed no-database run returned status: validated_no_load, three valid records, zero rejected records, and validated_total_usd: 100.00. That confirms parsing and arithmetic for this fixture. It does not confirm that any rows were written to PostgreSQL.

Each invocation creates its own directory under runs/, containing report.json and, after successful parsing, rejects.json. The report records an input SHA-256 hash, counts, UTC start time, and status. Rejections identify a logical CSV record number and reason, without copying the raw record into logs. Protect both input and report directories and define a retention policy for real data.

An exit code of 0 means validation succeeded or a load committed; read the report’s mode and status to distinguish them. Code 2 means rejected data blocked the load. Code 1 means a configuration, parsing, runtime, or report-write failure. A scheduler must observe these results; a log file alone does not notify anyone.

Keep validation separate from database writes

Python’s CSV reader returns text fields, so the application supplies the data contract. The sample rejects missing or extra columns, malformed records, invalid dates, unsupported amounts, and repeated IDs. Blank physical lines are skipped by the CSV reader; they are not counted as business records. A header-only file is an error.

The CLI only calls load_rows after the full input has been parsed with no rejected records. This is the relevant control flow from etl.py:

        rows, rejects, digest = extract_transform(args.csv_file)
        report.update(input_sha256=digest, valid_records=len(rows), rejected_records=len(rejects))
        (run / "rejects.json").write_text(json.dumps(rejects, indent=2), encoding="utf-8")
        if rejects:
            report["status"] = "rejected_no_load"
            exit_code = 2
        else:
            report["validated_total_usd"] = str(sum((r[2] for r in rows), Decimal("0.00")))
            if args.load:
                report["changed_records"] = load_rows(rows)
                report["status"] = "committed"
            else:
                report["status"] = "validated_no_load"
            exit_code = 0

The tests confirm that rejected data and a validation-only invocation never call the loader. They also check exact decimal totals, byte and record limits, encoding, quoted fields, a UTF-8 byte-order mark, duplicate IDs, and deterministic repeat validation. No test substitutes an in-memory database for PostgreSQL or claims that a mocked transaction proves rollback.

Prepare a disposable PostgreSQL table

Complete the earlier Compose lab’s authenticated application-access checks first. Its application role owns appdb and the app schema. Use that role rather than the database administrator. The role is intentionally broad for a development lab; a production runtime would usually have narrower grants and separate migration ownership.

In the lab’s authenticated psql session, execute the contents of schema.sql once:

-- Run once against an isolated database with an existing app schema owned by your lab role.
-- Existing table is an error; inspect it instead of dropping or truncating it.
CREATE TABLE app.etl_demo_orders (
    order_id varchar(40) PRIMARY KEY,
    order_date date NOT NULL,
    amount_usd numeric(10,2) NOT NULL CHECK (amount_usd >= 0)
);

An existing table is an error here. Inspect it rather than dropping, truncating, or silently reusing it. The primary key prevents duplicate stored IDs, while the numeric type and constraint enforce part of the data contract at the database boundary. Application validation does not replace database constraints.

Load the file in one transaction

The sample pins psycopg[binary]==3.3.8, a released client version checked on PyPI. It was installed and imported locally with Python 3.12.14. This is not a full dependency lock or proof of Linux image compatibility. The optional Docker build installs the same direct pin; validate and record the complete build before deployment.

The load uses parameterized values and a fixed table name. PostgreSQL’s ON CONFLICT behavior lets an existing order ID update its date and amount. The WHERE condition skips updates when those values are unchanged:

UPSERT = """
INSERT INTO app.etl_demo_orders AS target (order_id, order_date, amount_usd)
VALUES (%s, %s, %s)
ON CONFLICT (order_id) DO UPDATE SET
  order_date = EXCLUDED.order_date,
  amount_usd = EXCLUDED.amount_usd
WHERE (target.order_date, target.amount_usd)
  IS DISTINCT FROM (EXCLUDED.order_date, EXCLUDED.amount_usd)
"""

The loader wraps the batch in a Psycopg connection context. Its transaction documentation describes committing a successful context and rolling back when an exception escapes. A transaction-scoped advisory lock asks other copies of this same importer to stay out of the load:

def load_rows(rows):
    import psycopg
    # Psycopg's connection context commits normally and rolls back on an exception.
    with psycopg.connect(**connection_settings()) as conn:
        with conn.cursor() as cur:
            cur.execute("SET LOCAL statement_timeout = '30s'")
            cur.execute("SET LOCAL lock_timeout = '5s'")
            cur.execute("SELECT pg_try_advisory_xact_lock(17018301, 1)")
            if not cur.fetchone()[0]:
                raise RuntimeError("another cooperating import is active")
            cur.executemany(UPSERT, rows)
            changed = cur.rowcount
    return changed  # Reached only after the context manager commits successfully.

The advisory lock coordinates only clients using this same key in the same database. It does not block arbitrary SQL writers. The 30-second statement timeout and five-second lock timeout are example per-operation limits, not a deadline for the entire batch. No lock-race or timeout test was run against PostgreSQL.

The connection configuration requires PGHOST, PGPORT, PGDATABASE, PGUSER, and PGPASSFILE. Use a protected libpq password file, not a password in a shell command or connection URL. Verify TLS for connections that cross a network requiring it. The private Compose adapter below does not configure database TLS.

Connect the job to the existing private Compose lab

The following container steps are untested. The earlier database service has no published port. Its hostname db is reachable from services on its Compose network, not automatically from a host terminal or your laptop. Do not expose port 5432 to make this example work.

Put the extracted sample at pg-compose-lab/etl/, beside the existing compose.yaml. Create etl/runs with access limited to the operator. In a protected editor, create secrets/etl_pgpass with a line for db:5432:appdb:app and the lab application password. Follow libpq’s escaping rules for colons and backslashes. Keep the existing secrets-directory exclusion in Git and do not include credentials in the sample archive.

The supplied etl/compose.etl.yaml adds a one-shot job to the existing private database network:

services:
  etl:
    profiles: [etl]
    build:
      context: ./etl
    environment:
      PGHOST: db
      PGPORT: "5432"
      PGDATABASE: appdb
      PGUSER: app
      PGPASSFILE: /tmp/etl.pgpass
    secrets: [etl_pgpass]
    volumes:
      - ./etl/orders.csv:/app/orders.csv:ro
      - ./etl/runs:/app/runs
    networks: [database]
    depends_on:
      db:
        condition: service_healthy
    entrypoint: [sh, -c]
    command:
      - 'umask 077; cp /run/secrets/etl_pgpass /tmp/etl.pgpass && chmod 600 /tmp/etl.pgpass && exec python etl.py orders.csv --load --runs-dir /app/runs'
secrets:
  etl_pgpass:
    file: ./secrets/etl_pgpass

The job copies the password file into its temporary filesystem with mode 0600, as libpq requires on Unix. The teaching image uses its default root user, has no privileged flag, and does not mount the Docker socket. A non-root runtime identity and host-directory ownership need their own configuration and test before real deployment. The source CSV is mounted read-only; only the reports directory is writable through a bind mount.

From pg-compose-lab, run each proposed command only after the preceding step succeeds:

docker compose -f compose.yaml -f etl/compose.etl.yaml --profile etl config --quiet
docker compose -f compose.yaml -f etl/compose.etl.yaml build etl
docker compose -f compose.yaml -f etl/compose.etl.yaml run --rm etl

The merged configuration passed a local check with Docker Compose 5.1.4. Build, image pull, startup, secret permissions, database access, and the job were not executed because the Linux engine was unavailable. A healthy database probe is not evidence of correct application credentials. See Compose networking for service-name scope. The Python base-image tag can change; record the digest of a build you have tested.

Verify reruns, changes, and failures against the database

After a successful load into the initially empty lab table, use the authenticated psql client from the Compose guide:

SELECT order_id, order_date, amount_usd
FROM app.etl_demo_orders ORDER BY order_id;

SELECT count(*), sum(amount_usd) FROM app.etl_demo_orders;

Expected first result: the three fixture records. Expected totals: three rows and 100.00 USD. Run the identical file again and verify that the stored contents are unchanged and the report shows changed_records: 0. These expectations have not been checked against a PostgreSQL server here.

Change ORD-1002 from 80.00 to 81.00 and test again: expect the same three IDs, a total of 101.00, and one changed record. This is update-by-business-key behavior. Replaying an older file can overwrite a newer correction, and rows absent from the file are not deleted. The sample has no source-version ordering, change-data-capture stream, or exactly-once delivery guarantee.

For a rollback exercise in a disposable database, deliberately add a temporary constraint that rejects a later record while allowing an earlier one. Run the valid CSV and confirm that the failure leaves neither partial insertions nor partial updates. Restore the test schema only after inspecting the outcome. Also test wrong credentials, an unavailable server, a competing importer, and a lost connection during commit. Record the actual outcome of each check.

A connection failure during commit can leave the client unsure whether PostgreSQL committed. The JSON report is also written after the database transaction; a report-write failure cannot undo a commit. Reconcile the stored rows before retrying. The sample does not implement an atomic audit ledger, automatic retries, or recovery orchestration.

Once you have verified the imported rows and totals, use the Python PDF report guide to turn a separate, validated JSON input into a readable report. The PDF sample does not query this PostgreSQL table or consume the ETL run report directly; any export and field mapping need their own implementation and checks. The PDF guide records its local rendering tests, while this ETL-to-PDF integration, scheduling, and cloud execution remain untested.

Schedule the import after the manual checks pass

One small daily import can use the host’s scheduler. Supply a complete, stable input file before starting: the producer should close a temporary file and atomically rename it into place on the same filesystem. Reading while another process is writing can produce an incomplete batch even when individual records are valid.

Use one scheduling owner, explicit working directories and credentials, and a defined time zone. On a Linux host whose cron daemon is verified to use UTC, a proposed 02:15 daily entry could invoke an operator-owned wrapper:

15 2 * * * /usr/bin/flock -n -E 75 /opt/etl-lab/import.lock /opt/etl-lab/run-import.sh >> /opt/etl-lab/import.log 2>&1

This entry was not installed or tested. Verify the installed cron implementation, command paths, timezone, file permissions, and flock availability. The wrapper must change into the lab directory, run the tested job command, and preserve its exit code. Treat lock-conflict exit 75 and every nonzero job result as work to investigate. A daily cron entry does not automatically replay missed runs after downtime; define catch-up and file-selection rules explicitly.

Use the existing Celery worker guide when multiple producers or independently queued jobs justify a worker system. It carries its own untested runtime limits. Adding a queue does not make a non-idempotent import safe. If work runs concurrently, use the worker concurrency guide to account for database connections, memory, and downstream limits.

Choose where the ETL job should run

A local script may be enough for an occasional import. A managed scheduled-job service may be preferable when your team does not want to maintain a host. A CPU VM becomes useful when you need a recurring Linux environment, control over dependencies and networking, or several related tasks on one machine. Compare the operating work as well as the hourly compute price.

Use the CPU VM buying guide to assess that choice. Hivenet’s vCPU documentation includes non-GPU data-processing scripts and development/test databases. That supports an evaluation of this workload; it does not establish a managed ETL product, production database suitability, available capacity, or a tested Hivenet deployment.

Measure a representative batch’s runtime, peak memory, database behavior, and accepted output before changing VM size. Include idle running time and operational effort in cost comparisons. Keep source files and recovery copies outside the VM’s failure boundary: Hivenet’s lifecycle guidance says stopped instances expire after ten days and local data is deleted. A container volume does not override that rule.

The next useful result is a repeatable import with recorded failure and rerun checks. Keep the local validation evidence separate from database, scheduler, recovery, and cloud evidence, then choose capacity from the workload you have actually measured.