Chapter 4 · Python and Data Handling
Importing and Exporting: CSV, JSON, Excel and Parquet
- Page 16 of 22
- 18 min read
Data rarely starts or ends in your database. It arrives as a CSV from a partner, an Excel sheet from finance, JSON from an API, a Parquet file from the data lake, and it leaves the same ways: a report for a manager, a JSON Lines file of examples for fine-tuning an LLM, a Parquet snapshot of training data. Every one of these crossings can corrupt data quietly: a broken encoding turns Bangla names into garbage, a comma inside an address shifts every column, one bad row aborts a load of a million.
This page is the practical companion to the format comparison in Data Handling Fundamentals: Formats, Quality and Lineage: how to read and write each format correctly from Python, how to load files into a database in bulk, and how to keep bad rows out without losing them.
What you will learn
- CSV done right: quoting, delimiters, encodings and UTF-8 for Bangla text.
- Loading files into a database (csv module, pandas, staging tables, PostgreSQL COPY, MySQL LOAD DATA) with validation and a quarantine for failed rows.
- Exporting query results to CSV automatically, with a manifest.
- Excel with pandas and openpyxl; JSON, nested JSON (
json_normalize) and JSON Lines; XML basics. - Parquet and columnar storage, file sizes compared, and processing big files in chunks.
CSV: quoting, delimiters and encodings
CSV looks trivial and is not. Let Python's csv module write a file that contains the two classic traps, a comma inside a value and a quote inside a value, plus a Bangla name:
import csv
new_customers = [
{"customer_id": 9, "name": "Rumana Begum", "email": "rumana@example.com", "city": "Barishal", "joined_on": "2026-06-15", "address": "House 12, Road 5, Dhanmondi"},
{"customer_id": 10, "name": "তানিয়া সুলতানা", "email": "tania@example.com", "city": "Dhaka", "joined_on": "2026-06-18", "address": "Shop \"Tech Corner\", Level 3"},
{"customer_id": 11, "name": "Arif Hasan", "email": "nadia@example.com", "city": "Sylhet", "joined_on": "2026-06-20", "address": "Zindabazar"},
{"customer_id": 12, "name": "Jui Akter", "email": "jui@example.com", "city": "Khulna", "joined_on": "2026-13-01", "address": "Sonadanga"},
{"customer_id": 13, "name": "", "email": "blank@example.com", "city": "Dhaka", "joined_on": "2026-06-22", "address": "Mirpur 10"},
{"customer_id": 14, "name": "Sabbir Khan", "email": "sabbir@example.com", "city": "", "joined_on": "2026-06-25", "address": "Agrabad"},
]
with open("new_customers.csv", "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=list(new_customers[0]))
writer.writeheader()
writer.writerows(new_customers)
with open("new_customers.csv", encoding="utf-8") as f:
print(f.read())customer_id,name,email,city,joined_on,address
9,Rumana Begum,rumana@example.com,Barishal,2026-06-15,"House 12, Road 5, Dhanmondi"
10,তানিয়া সুলতানা,tania@example.com,Dhaka,2026-06-18,"Shop ""Tech Corner"", Level 3"
11,Arif Hasan,nadia@example.com,Sylhet,2026-06-20,Zindabazar
12,Jui Akter,jui@example.com,Khulna,2026-13-01,Sonadanga
13,,blank@example.com,Dhaka,2026-06-22,Mirpur 10
14,Sabbir Khan,sabbir@example.com,,2026-06-25,AgrabadThe writer put quotes around the address with a comma, and doubled the quotes inside "Tech Corner". That is the CSV standard (RFC 4180), and it is why you never split CSV lines yourself:
with open("new_customers.csv", encoding="utf-8") as f:
line = f.readlines()[1]
print(len(line.split(",")), "pieces with split(',')")
print(len(next(csv.reader([line]))), "fields with csv.reader")8 pieces with split(',')
6 fields with csv.readerOther delimiters follow the same rules: csv.reader(f, delimiter=";") for European exports (where the comma is the decimal separator) and delimiter="\t" for TSV; in pandas, pd.read_csv(path, sep=";"). Always pass newline="" when opening a CSV for the csv module, or quoted values containing line breaks get mangled.
Encodings: why Bangla turns into garbage
A file is bytes; an encoding says how characters become bytes. UTF-8 stores English letters in 1 byte and Bangla letters in 3. Read UTF-8 bytes with the wrong encoding and you get mojibake, or an error:
city = "ঢাকা"
data = city.encode("utf-8")
print(len(city), "characters,", len(data), "bytes")
print(data.decode("cp1252", errors="replace")) # read with the wrong encoding
try:
with open("new_customers.csv", encoding="ascii") as f:
f.read()
except UnicodeDecodeError as exc:
print("UnicodeDecodeError:", exc)4 characters, 12 bytes
ঢাকা
UnicodeDecodeError: 'ascii' codec can't decode byte 0xe0 in position 135: ordinal not in range(128)Rules that prevent almost every encoding bug:
- Always say
encoding="utf-8"when you open a text file. The default depends on the operating system (Windows often uses cp1252), so code that works on your laptop breaks on a colleague's. Python 3.15 makes UTF-8 the default (PEP 686), but code that also runs on older versions still needs it spelled out. - Excel on Windows guesses the encoding of a CSV and often guesses wrong. Write files meant for Excel with
encoding="utf-8-sig": it adds a 3-byte marker (the BOM) that tells Excel "this is UTF-8". - If a partner sends a file in an unknown encoding, ask. Libraries such as
charset-normalizercan guess, but a guess is not a fact.
Importing a CSV into the database, safely
Only three of the six rows above are fine. Row 11 reuses Nadia's email (the UNIQUE constraint will refuse it), row 12 has month 13, and row 13 has no name. A careful import validates each row, loads the good ones in one transaction, and writes the rejected ones with a reason to a quarantine file instead of failing the whole load or silently dropping them:
import sqlite3
from datetime import date
con = sqlite3.connect("shop.db")
con.execute("PRAGMA foreign_keys = ON")
def problem_with(row):
"""Return why a row is invalid, or None if it looks fine."""
if not row["name"].strip():
return "name is empty"
if "@" not in row["email"]:
return "email is invalid"
try:
date.fromisoformat(row["joined_on"])
except ValueError:
return "joined_on is not a valid date"
return None
loaded, quarantined = 0, []
with open("new_customers.csv", newline="", encoding="utf-8") as f, con:
for row in csv.DictReader(f):
reason = problem_with(row)
if reason is None:
try:
con.execute(
"INSERT INTO customers (customer_id, name, email, city, joined_on) VALUES (?, ?, ?, ?, ?)",
(int(row["customer_id"]), row["name"], row["email"], row["city"] or None, row["joined_on"]),
)
loaded += 1
continue
except sqlite3.IntegrityError as exc:
reason = f"database refused: {exc}"
quarantined.append({**row, "reason": reason})
with open("customers_quarantine.csv", "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=list(quarantined[0]))
writer.writeheader()
writer.writerows(quarantined)
print(loaded, "loaded,", len(quarantined), "quarantined")
for q in quarantined:
print(q["customer_id"], "-", q["reason"])3 loaded, 3 quarantined
11 - database refused: UNIQUE constraint failed: customers.email
12 - joined_on is not a valid date
13 - name is emptyValidate in Python what you can (formats, required fields), and let the database enforce what only it knows (uniqueness, foreign keys). Two details: an empty city becomes None (NULL) rather than an empty string, and a failed INSERT in SQLite undoes only that statement, so the transaction carries on. In PostgreSQL a failed statement aborts the whole transaction, as you saw in Python and Databases: Connections, Parameters and Transactions: wrap each row in a savepoint there, or validate fully before loading.
The staging-table pattern
For larger files, load everything first into a staging table with no constraints, then let SQL move the valid rows into the real table. The database does the heavy lifting, and the staging table keeps the raw rows for inspection:
import pandas as pd
staging = pd.read_csv("new_customers.csv", dtype=str, keep_default_na=False)
staging.to_sql("staging_customers", con, if_exists="replace", index=False)
with con:
cur = con.execute("""
INSERT INTO customers (customer_id, name, email, city, joined_on)
SELECT CAST(customer_id AS INTEGER), name, email, NULLIF(city, ''), joined_on
FROM staging_customers s
WHERE name <> ''
AND date(joined_on) = joined_on -- a real date
AND NOT EXISTS (SELECT 1 FROM customers c
WHERE c.email = s.email OR c.customer_id = CAST(s.customer_id AS INTEGER))
""")
print(cur.rowcount, "new rows (the earlier import already loaded the valid ones)")
print(pd.read_sql("SELECT customer_id, name, city FROM customers WHERE customer_id >= 9 ORDER BY customer_id", con))0 new rows (the earlier import already loaded the valid ones)
customer_id name city
0 9 Rumana Begum Barishal
1 10 তানিয়া সুলতানা Dhaka
2 14 Sabbir Khan NaNZero rows: the valid customers are already there, and the NOT EXISTS condition makes the load safe to repeat. That property, running twice changes nothing, is called idempotency, and it is the heart of the next page. Notice the Bangla name stored and read back intact.
Bulk loading on the servers
Row-by-row INSERTs, or one executemany per batch (see Python and Databases: Connections, Parameters and Transactions), are fine for thousands of rows. For millions, use the database's bulk loader, which reads a whole file in one command and is often 10 to 100 times faster. In PostgreSQL that is COPY. COPY table FROM '/path/file.csv' reads a file on the database server and needs superuser or the pg_read_server_files role; for a file on your own machine, use COPY … FROM STDIN and send the file over the connection. That is what psql's \copy command does, and what psycopg2's copy_expert does here (in psycopg 3 it is cur.copy(...)):
import os
import psycopg2
pg = psycopg2.connect(host=os.environ["PGHOST"], port=os.environ["PGPORT"], user=os.environ["PGUSER"],
password=os.environ["PGPASSWORD"], dbname=os.environ["PGDATABASE"])
with pg, pg.cursor() as cur:
cur.execute("CREATE TABLE staging_customers (customer_id text, name text, email text, "
"city text, joined_on text, address text)")
with open("new_customers.csv", encoding="utf-8") as f:
cur.copy_expert("COPY staging_customers FROM STDIN WITH (FORMAT csv, HEADER true)", f)
cur.execute("SELECT customer_id, name, address FROM staging_customers ORDER BY customer_id::int LIMIT 2")
for row in cur.fetchall():
print(row)
pg.close()('9', 'Rumana Begum', 'House 12, Road 5, Dhanmondi')
('10', 'তানিয়া সুলতানা', 'Shop "Tech Corner", Level 3')COPY understood the quoting and the UTF-8 by itself. MySQL's equivalent is LOAD DATA. Without LOCAL it reads a file on the server, limited to the folder named by secure_file_priv. With LOCAL it reads a file from the client machine, which needs local_infile switched on in both the server and the client (for example local_infile=True in pymysql.connect); it is off by default in MySQL 8 for security, so it is shown here but not run:
-- MySQL
LOAD DATA LOCAL INFILE 'new_customers.csv'
INTO TABLE staging_customers
CHARACTER SET utf8mb4
FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '"'
LINES TERMINATED BY '\r\n'
IGNORE 1 LINES;
csv.writerends lines with\r\nby default (the CSV standard), which is why the MySQL example says so. A loader expecting\nwould keep a stray\rat the end of the last column of every row.
Exporting: SQL to CSV, automatically
An export that runs every night should be boring and checkable. Stream rows from a cursor straight into the file (no need to hold them all in memory), and write a small manifest next to it with the row count and a checksum, so whoever receives it can verify nothing was lost or changed:
import hashlib
import json
def export_query(con, sql, params, path):
"""Write a query's rows to a UTF-8 CSV plus a JSON manifest; return the manifest."""
cur = con.execute(sql, params)
rows = 0
with open(path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow([d[0] for d in cur.description])
for row in cur:
writer.writerow(row)
rows += 1
with open(path, "rb") as f:
digest = hashlib.sha256(f.read()).hexdigest()
manifest = {"file": path, "rows": rows, "sha256": digest[:16]}
with open(path + ".manifest.json", "w", encoding="utf-8") as f:
json.dump(manifest, f)
return manifest
print(export_query(
con,
"SELECT order_id, customer_id, order_date, status FROM orders WHERE order_date >= ? ORDER BY order_id",
("2026-05-01",),
"orders_2026-05.csv",
)){'file': 'orders_2026-05.csv', 'rows': 4, 'sha256': '5993311f728321ff'}Put the period in the file name (orders_2026-05.csv), never overwrite yesterday's file with today's, and schedule the script with cron or an orchestrator (next page). With pandas the export is one line, pd.read_sql(sql, con).to_csv(path, index=False), at the cost of holding the result in memory.
Excel
pandas reads and writes .xlsx through openpyxl. One workbook can hold several sheets, and sheet_name=None reads them all into a dictionary:
orders = pd.read_sql("SELECT order_id, customer_id, order_date, status FROM orders ORDER BY order_id", con)
products = pd.read_sql("SELECT product_id, name, price FROM products ORDER BY product_id", con)
with pd.ExcelWriter("shop_report.xlsx") as xl:
orders.to_excel(xl, sheet_name="orders", index=False)
products.to_excel(xl, sheet_name="products", index=False)
sheets = pd.read_excel("shop_report.xlsx", sheet_name=None)
print({name: df.shape for name, df in sheets.items()}){'orders': (14, 4), 'products': (11, 3)}For formatting, or for reading a sheet that is not a clean table (a title row, merged cells, totals at the bottom), use openpyxl directly:
from openpyxl import load_workbook
from openpyxl.styles import Font
wb = load_workbook("shop_report.xlsx")
ws = wb["products"]
for cell in ws[1]:
cell.font = Font(bold=True) # bold header row
ws.column_dimensions["B"].width = 30
ws.freeze_panes = "A2" # keep the header visible when scrolling
wb.save("shop_report.xlsx")
for row in ws.iter_rows(min_row=1, max_row=3, values_only=True):
print(row)('product_id', 'name', 'price')
(1, 'Python Crash Course', 1200)
(2, 'Hands-On Machine Learning', 2500)Excel files carry traps of their own: dates stored as numbers, numbers stored as text, leading zeros dropped from phone numbers, and a limit of 1,048,576 rows. Treat Excel as a format for people, not as a pipeline's storage.
JSON, nested JSON and JSON Lines
APIs return JSON, and it is usually nested: an order holds a customer object and a list of items. A table needs one row per item. pd.json_normalize flattens it: record_path names the list that becomes rows, meta names the parent fields to copy onto each row. Here a local file stands in for the API response:
api_response = {
"orders": [
{"order_id": 6, "customer": {"id": 2, "name": "Tanvir Ahmed"},
"items": [{"sku": "AI-BOOT", "qty": 1, "price": 15000}]},
{"order_id": 9, "customer": {"id": 6, "name": "Imran Hossain"},
"items": [{"sku": "KEYB-MECH", "qty": 1, "price": 4050},
{"sku": "MOUSE-WL", "qty": 1, "price": 900}]},
]
}
with open("orders_api.json", "w", encoding="utf-8") as f:
json.dump(api_response, f)
with open("orders_api.json", encoding="utf-8") as f:
payload = json.load(f)
items = pd.json_normalize(payload["orders"], record_path="items",
meta=["order_id", ["customer", "name"]])
print(items) sku qty price order_id customer.name
0 AI-BOOT 1 15000 6 Tanvir Ahmed
1 KEYB-MECH 1 4050 9 Imran Hossain
2 MOUSE-WL 1 900 9 Imran HossainGoing the other way, JSON Lines (one JSON object per line, .jsonl) is the standard format for LLM fine-tuning data, evaluation sets and logs, because it can be appended to and streamed line by line. Two details matter for Bangla text:
examples = pd.DataFrame({
"prompt": ["Where does customer 10 live?", "What is the name of customer 10?"],
"answer": ["Dhaka", "তানিয়া সুলতানা"],
})
print(json.dumps({"answer": "তানিয়া"})) # default: escaped
print(json.dumps({"answer": "তানিয়া"}, ensure_ascii=False)) # readable UTF-8
examples.to_json("examples.jsonl", orient="records", lines=True, force_ascii=False)
with open("examples.jsonl", encoding="utf-8") as f:
print(f.read().strip())
print(pd.read_json("examples.jsonl", lines=True).shape){"answer": "\u09a4\u09be\u09a8\u09bf\u09af\u09bc\u09be"}
{"answer": "তানিয়া"}
{"prompt":"Where does customer 10 live?","answer":"Dhaka"}
{"prompt":"What is the name of customer 10?","answer":"তানিয়া সুলতানা"}
(2, 2)Escaped JSON is still correct (any parser turns \u09a4 back into ত), but it is unreadable to people and about twice as large on disk. Use ensure_ascii=False (json) or force_ascii=False (pandas) and write the file as UTF-8.
XML basics
XML still arrives from banks, government systems and older enterprise software. For simple, repeated records, pd.read_xml turns each element into a row (attributes and child elements become columns); xml.etree.ElementTree from the standard library handles anything less regular:
import xml.etree.ElementTree as ET
xml_text = """<?xml version="1.0" encoding="UTF-8"?>
<price_feed supplier="Gadget House">
<item sku="KEYB-MECH"><name>Mechanical Keyboard</name><price>4300</price></item>
<item sku="MOUSE-WL"><name>Wireless Mouse</name><price>850</price></item>
</price_feed>"""
with open("price_feed.xml", "w", encoding="utf-8") as f:
f.write(xml_text)
root = ET.parse("price_feed.xml").getroot()
print(root.get("supplier"))
for item in root.iter("item"):
print(item.get("sku"), item.findtext("name"), int(item.findtext("price")))
print(pd.read_xml("price_feed.xml", xpath=".//item", parser="etree"))Gadget House
KEYB-MECH Mechanical Keyboard 4300
MOUSE-WL Wireless Mouse 850
sku name price
0 KEYB-MECH Mechanical Keyboard 4300
1 MOUSE-WL Wireless Mouse 850parser="etree" uses the standard library; the default, "lxml", is faster but is a separate install. Never parse XML from an untrusted source with these without care: crafted XML can exhaust memory. The defusedxml package is the safe drop-in.
Parquet: columnar storage
CSV and JSON store data row by row as text. Parquet stores it column by column in binary, with the types written into the file and each column compressed separately:
Row format (CSV) Columnar format (Parquet)
1,2026-01-12,chat,312 event_id: 1, 2, 3, ... (int64)
2,2026-01-12,view,45 day: 2026-01-12, ... (date)
3,2026-01-13,chat,280 kind: chat, view, ... (few distinct values: compresses well)
... seconds: 312, 45, 280, ... (int64)That brings three benefits: smaller files (similar values sit together and compress well), faster reads of a few columns (a query that needs seconds never reads kind), and types that survive the round trip. pandas reads and writes Parquet through pyarrow, a separate install (pip install pyarrow). Compare with 200,000 generated events:
import numpy as np
rng = np.random.default_rng(0)
n = 200_000
events = pd.DataFrame({
"event_id": np.arange(1, n + 1),
"day": pd.to_datetime("2026-01-01") + pd.to_timedelta(rng.integers(0, 180, n), unit="D"),
"kind": rng.choice(["view", "search", "add_to_cart", "chat"], n),
"seconds": rng.integers(1, 600, n),
})
events.to_csv("events.csv", index=False)
events.to_parquet("events.parquet", index=False)
for path in ["events.csv", "events.parquet"]:
print(f"{path:15} {os.path.getsize(path) / 1_000_000:5.1f} MB")
back_csv = pd.read_csv("events.csv")
back_parquet = pd.read_parquet("events.parquet", columns=["day", "seconds"]) # read only two columns
print("day from CSV:", back_csv["day"].dtype, "| day from Parquet:", back_parquet["day"].dtype)events.csv 5.7 MB
events.parquet 1.6 MB
day from CSV: str | day from Parquet: datetime64[us]The CSV lost the date type (it came back as text), Parquet kept it. Parquet is the standard file format of data lakes, Spark, DuckDB and warehouse exports, and the right choice for saving training datasets. CSV remains the right choice for small files that people open, and for systems that only speak CSV.
Big files: chunks and streaming
When a file is larger than memory, never load it whole. read_csv(chunksize=…) yields DataFrames of a fixed size; aggregate each and combine:
totals = {}
for chunk in pd.read_csv("events.csv", usecols=["kind", "seconds"], chunksize=50_000):
for kind, s in chunk.groupby("kind")["seconds"].sum().items():
totals[kind] = totals.get(kind, 0) + int(s)
print(totals)
def stream_rows(path):
"""Yield one dict per CSV row: constant memory, however big the file."""
with open(path, newline="", encoding="utf-8") as f:
yield from csv.DictReader(f)
long_chats = sum(1 for row in stream_rows("events.csv") if row["kind"] == "chat" and int(row["seconds"]) > 500)
print(long_chats, "chats longer than 500 seconds"){'add_to_cart': 14949553, 'chat': 15001659, 'search': 15148999, 'view': 14898307}
8295 chats longer than 500 secondsChunks keep pandas' speed with bounded memory; the generator uses almost no memory at all and suits row-at-a-time work such as validating and loading. For analytical queries over big CSV or Parquet files, DuckDB can run SQL on them directly without loading them into a database first.
Common mistakes
- Opening files without
encoding="utf-8". It works on Linux and breaks on Windows. Always name the encoding, and useutf-8-sigfor CSVs that people will open in Excel. - Splitting CSV lines with
split(","). The first address with a comma breaks every column after it. Usecsvorpandas. - Letting one bad row fail (or silently shrink) a load. Validate, load the good rows, quarantine the rest with a reason, and report both counts.
- Letting pandas guess types on import. IDs like
"0172…"lose their leading zero, codes like"NA"become missing values. Read withdtype=str(andkeep_default_na=False) for anything that is an identifier, then convert on purpose. - Using CSV to store training data between pipeline steps. Types are lost and must be re-parsed on every read. Use Parquet.
Try it yourself
- Easy: export the products table to
products_for_excel.csvwith pandas so that Excel opens it correctly, then check that the file starts with the UTF-8 BOM bytesb'\xef\xbb\xbf'. - Medium: write the delivered orders with their items as nested JSON (one object per order with an
itemslist) todelivered.json, then flatten it back withjson_normalizeand check that the row count equals the number of order_items rows of delivered orders. - Hard: write
load_csv(con, path, table, required)that loads any CSV into an existing table with thecsvmodule, quarantines rows with an empty required field or a database error to<path>.rejected.csv, and returns(loaded, rejected). Test it by loadingprice_changes.csv(write it yourself: three rows for a new tableprice_changeswithproduct_id(a foreign key to products) andnew_price INTEGER CHECK (new_price > 0), one with product 99 and one with price 0).
Answers
# 1. Easy
pd.read_sql("SELECT * FROM products ORDER BY product_id", con).to_csv(
"products_for_excel.csv", index=False, encoding="utf-8-sig")
with open("products_for_excel.csv", "rb") as f:
print(f.read(3))
# 2. Medium
lines = pd.read_sql("""
SELECT o.order_id, o.order_date, oi.product_id, oi.quantity, oi.unit_price
FROM orders o JOIN order_items oi ON oi.order_id = o.order_id
WHERE o.status = 'delivered' ORDER BY o.order_id, oi.product_id""", con)
nested = [
{"order_id": int(oid), "order_date": g["order_date"].iloc[0],
"items": g[["product_id", "quantity", "unit_price"]].to_dict("records")}
for oid, g in lines.groupby("order_id")
]
with open("delivered.json", "w", encoding="utf-8") as f:
json.dump(nested, f, default=int)
with open("delivered.json", encoding="utf-8") as f:
flat = pd.json_normalize(json.load(f), record_path="items", meta=["order_id", "order_date"])
print(len(flat), len(lines), len(flat) == len(lines))
# 3. Hard
def load_csv(con, path, table, required):
loaded, rejected = 0, []
with open(path, newline="", encoding="utf-8") as f, con:
reader = csv.DictReader(f)
cols = reader.fieldnames
sql = f"INSERT INTO {table} ({', '.join(cols)}) VALUES ({', '.join('?' * len(cols))})"
for row in reader:
empty = [c for c in required if not row[c].strip()]
if empty:
rejected.append({**row, "reason": f"empty: {', '.join(empty)}"})
continue
try:
con.execute(sql, [row[c] or None for c in cols])
loaded += 1
except sqlite3.Error as exc:
rejected.append({**row, "reason": str(exc)})
if rejected:
with open(path + ".rejected.csv", "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=list(rejected[0]))
w.writeheader()
w.writerows(rejected)
return loaded, len(rejected)
con.execute("CREATE TABLE price_changes (product_id INTEGER, new_price INTEGER CHECK (new_price > 0), "
"FOREIGN KEY (product_id) REFERENCES products (product_id))")
with open("price_changes.csv", "w", newline="", encoding="utf-8") as f:
f.write("product_id,new_price\n3,4300\n99,500\n4,0\n")
print(load_csv(con, "price_changes.csv", "price_changes", required=["product_id", "new_price"]))
with open("price_changes.csv.rejected.csv", encoding="utf-8") as f:
print(f.read().strip())Summary
- Use the
csvmodule or pandas, neversplit(","); always open text files withencoding="utf-8"(andutf-8-sigfor Excel). Bangla text survives every step when the encoding is explicit. - Import with validation, one transaction for the good rows, and a quarantine file with reasons for the rest; staging tables let SQL do the checking; COPY and LOAD DATA are for bulk volumes.
- Automated exports stream from a cursor and write a manifest (row count, checksum).
- Excel is for people; JSON needs
json_normalizewhen nested; JSON Lines is the format of LLM datasets; XML parses withread_xmlor ElementTree. - Parquet is columnar, typed and compressed: the right format for datasets. Process big files in chunks or as a stream.
Next: ETL and ELT: Building a Reliable Data Pipeline joins everything from this chapter (connections, pandas, cleaning, files) into one pipeline that extracts from a database, an API and a file, transforms, loads, and can be run again and again without changing a thing.