Importing and Exporting Data

Every data layer eventually moves data in and out: a CSV from the legacy system, a JSON dump for the backup or the API, a migration from the old schema to the new. The import is a read of the external format into rows and a bulk write of those rows (the bulk chapter's insert_many, chunked); the export is a query and a serialization of the result. This chapter covers both directions with the file formats the app actually meets - CSV and JSON - the validation that belongs at the import boundary, the idempotency that makes a re-run safe, and the export patterns (the full dump, the filtered extract, the pagination for the large one).

Importing a CSV

import csv
from peewee import *

def import_authors(path):
    # the import is three steps: read the file, validate each row,
    # bulk-write the valid ones (chunked, in a transaction)
    rows = []
    with open(path, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)     # the header names the columns
        for raw in reader:
            # the validation belongs HERE, at the boundary:
            # a bad row is rejected (and logged), not crashed on
            if not raw.get("email"):
                log.warning("missing email, skipping: %r", raw)
                continue
            rows.append({
                "name": raw["name"].strip(),
                "email": raw["email"].strip().lower(),
                "bio": raw.get("bio") or None,
            })

    # the bulk write: chunked insert_many in ONE transaction,
    # so a failure rolls back the whole import
    with atomic(db):
        for i in range(0, len(rows), 1000):
            Author.insert_many(rows[i:i + 1000]).execute()
    return len(rows)

The CSV import is the read-validate-write shape. The csv.DictReader maps the header to the field names, so each row is a dict keyed by the column - the same shape insert_many wants, which is why the file format and the bulk API meet cleanly. The validation is at the boundary, per row: a missing or malformed field is logged and skipped (the import continues, the bad rows are reported) rather than raising and aborting the whole file, because a legacy export with a hundred bad rows among a hundred thousand is the normal case, not the exception. The write is the bulk chapter's chunked insert_many wrapped in atomic - one transaction for the whole import, so a failure at row 50000 undoes the 49999 before it. The three steps - read, validate, bulk-write - are the structure every file import has, and the chunk-and-transaction are the scale and correctness layers over it.

Idempotency: The Re-Run Safety

from peewee import *

# a naive insert import fails on re-run: the unique email is
# already there. the fix is the upsert (the bulk chapter) or the
# skip-existing check, so RE-RUNNING the import is safe:

def import_authors_idempotent(rows):
    with atomic(db):
        for i in range(0, len(rows), 1000):
            batch = rows[i:i + 1000]
            (Author.insert_many(batch)
                     .on_conflict(
                         conflict_target=[Author.email],
                         preserve=[Author.id],
                         update={Author.name: Author.name},  # keep existing
                     )
                     .execute())

# the contract: running the import twice with the same file
# leaves the database in the same state as running it once -
# the existing rows are untouched (or updated, deliberately),
# the new rows are added, and the operation is safe to retry
# after a failure, which is what makes an unattended import
# (a scheduled job, a retrying worker) possible

The property that makes an import operable is idempotency: running it twice with the same input leaves the database as running it once did. The naive insert_many is not idempotent - the second run hits the unique constraint and fails - and the fix is the upsert form (on_conflict with the unique key as the target, preserving the existing row or updating it deliberately) or an explicit skip-existing check. The idempotent import is what an unattended job needs: the scheduled load, the retrying worker, the re-run after a partial failure - all of them assume the operation can be repeated without doubling the data, and the unique-key upsert is the mechanism that gives them that guarantee. The decision inside the upsert - preserve the existing value or update it - is the policy, and it should be the one the data owner stated ("the file is the truth" means update, "the database is the truth" means preserve), not a default the code guessed.

Exporting: The Query and the Serialization

import csv, json
from peewee import *

def export_articles(path, published_only=True):
    # the export is a query + a serialization, field by field:
    q = Article.select()
    if published_only:
        q = q.where(Article.published == True)
    with open(path, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["id", "title", "slug", "views", "author"])
        # the JOIN brings the related value into the row:
        for a in (q.join(Author)
                    .execute()):
            w.writerow([a.id, a.title, a.slug, a.views, a.author.name])

def export_articles_json(path):
    # the JSON form: the same query, serialized as objects
    rows = []
    for a in Article.select().join(Author).execute():
        rows.append({
            "id": a.id, "title": a.title, "slug": a.slug,
            "views": a.views, "author": a.author.name,
        })
    with open(path, "w", encoding="utf-8") as f:
        json.dump(rows, f, indent=2, default=str)
    # default=str handles the date/datetime fields the JSON
    # encoder does not know - the serialization's standard escape

The export is the mirror of the import: a query (the filtered select, with the join when the row needs a related value like the author's name) and a serialization to the file format, field by field. The CSV form writes the header and a row per instance, pulling the attributes in the column order; the JSON form builds a dict per row and json.dumps the list, with default=str as the standard handler for the date and Decimal fields the encoder does not natively serialize. The explicit field list is the point - the export names the columns it writes, which is both the documentation of the format and the protection against a model change silently adding a column to the file. The join in the export query is the denormalization the file format wants (the author's name in the row, not a foreign key the consumer must resolve), and it is the join chapter's tool doing the export's job.

The Large Export: Pagination

from peewee import *

def export_all(path, page=5000):
    # a full-table export of a large table is paged by the id:
    # the query takes a slice, writes it, advances the cursor
    written = 0
    last_id = 0
    with open(path, "w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["id", "title", "slug", "views"])
        while True:
            batch = (Article.select()
                         .where(Article.id > last_id)
                         .order_by(Article.id)
                         .limit(page)
                         .execute())
            if not batch:
                break
            for a in batch:
                w.writerow([a.id, a.title, a.slug, a.views])
                last_id = a.id
            written += len(batch)
    return written

# the keyset pagination (id > last_id) over the offset pagination
# (LIMIT/OFFSET) - the offset form re-scans the skipped rows on
# each page, which is the O(n^2) cost on a large table; the keyset
# form seeks to the cursor, which is the O(n) export

The large export is paged, because materializing the whole table in memory is not an option at the scale a full dump reaches. The paging is keyset (seek) pagination - WHERE id > last_id ORDER BY id LIMIT page, advancing the cursor to the last id written - rather than offset pagination (LIMIT page OFFSET n*page), because the offset form re-scans the skipped rows on every page and the cost grows with the table, while the keyset form seeks to the cursor in the index and the cost is linear in the rows written. The page size (5000) is the memory knob - large enough to amortize the query, small enough to bound the batch in memory - and the loop writes each page to the file and advances, so the file is built in streaming fashion and the peak memory is one page, not the table. The pattern is the standing one for any "read the whole table" operation - the export, the backfill, the report - and it is the offset-pagination trap the performance chapter flags, solved the same way here.

Summary

The import is read-validate-bulk-write: the DictReader maps the file's header to the fields, the validation rejects and logs the bad rows at the boundary (the import continues), and the chunked insert_many in one atomic block writes the good ones with the roll-back-on-failure guarantee. Idempotency - the re-run safety the unattended job needs - is the on_conflict upsert on the unique key, with the preserve-or-update policy the data owner stated. The export is the mirror: a filtered query (joined for the denormalized values the file wants) serialized field by field to CSV or JSON, with default=str for the date and Decimal fields, and the explicit column list as the format's documentation. The large export is keyset-paged (id > last_id), streaming the table in bounded pages instead of materializing it - the O(n) form over the offset's O(n-squared) re-scan. The file is the boundary where the external world and the model meet, and the validation, the idempotency, and the paging are what make that boundary safe at the scale the data actually is.