Before you start
- Python 3.9 or newer (the sqlite3 and csv modules ship with it)
- Basic SQL: SELECT, GROUP BY
- A terminal and a folder to work in
A pipeline is three jobs in a row: get data in, make sure it is fit to use, and answer a question with it. This tutorial builds all three in about sixty lines, using only what ships with Python.
You will generate a small export of shop orders, load it into SQLite while rejecting rows that are not valid, and report revenue by month with a running total.
Step 1: Create some data to work with
Real exports are messy, so this script writes 300 clean orders and then adds two rows of the kind you meet in practice: a blank quantity and a price with a currency sign.
import csv
import random
from datetime import date, timedelta
random.seed(7)
PRODUCTS = {"notebook": 4.50, "pen": 1.20, "backpack": 39.00, "lamp": 24.00}
start = date(2025, 1, 1)
with open("orders.csv", "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["order_id", "order_date", "customer_id", "product", "quantity", "unit_price"])
for order_id in range(1, 301):
day = start + timedelta(days=random.randint(0, 119))
product = random.choice(list(PRODUCTS))
writer.writerow([order_id, day.isoformat(), random.randint(1, 60), product, random.randint(1, 5), f"{PRODUCTS[product]:.2f}"])
# two rows a real export would contain
writer.writerow([301, "2025-03-02", 12, "lamp", "", "24.00"])
writer.writerow([302, "2025-03-03", 14, "pen", 2, "$1.20"])
print("wrote orders.csv")
Save it as generate.py and run it. The seed makes the data repeatable on your machine. Another Python version may shuffle the numbers slightly, but the shape of every result below stays the same.
Step 2: Load the file, validating as you go
The table definition does the first line of defence. NOT NULL and CHECK constraints mean the database itself refuses a nonsense row, whatever the script does.
import csv
import sqlite3
conn = sqlite3.connect("shop.db")
conn.execute("DROP TABLE IF EXISTS orders")
conn.execute(
"""
CREATE TABLE orders (
order_id INTEGER PRIMARY KEY,
order_date TEXT NOT NULL,
customer_id INTEGER NOT NULL,
product TEXT NOT NULL,
quantity INTEGER NOT NULL CHECK (quantity > 0),
unit_price REAL NOT NULL CHECK (unit_price >= 0)
)
"""
)
good, rejected = [], []
with open("orders.csv", newline="", encoding="utf-8") as f:
for row in csv.DictReader(f):
try:
good.append((
int(row["order_id"]),
row["order_date"].strip(),
int(row["customer_id"]),
row["product"].strip().lower(),
int(row["quantity"]),
float(row["unit_price"]),
))
except ValueError as error:
rejected.append((row["order_id"], str(error)))
with conn:
conn.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?)", good)
print(f"loaded {len(good)} rows, rejected {len(rejected)}")
for order_id, reason in rejected:
print(f" order {order_id}: {reason}")
Save it as load.py. Three details matter:
- Conversion is validation.
int("")andfloat("$1.20")raiseValueError, so the two bad rows are caught without any special-case code. with conn:is a transaction. Either every good row is saved or none is. A crash half-way leaves no half-loaded table.- Rejected rows are kept, not hidden. The script prints them, which is the list you send back to the owner of the export.
Run it and you should see:
loaded 300 rows, rejected 2
order 301: invalid literal for int() with base 10: ''
order 302: could not convert string to float: '$1.20'
Step 3: Ask a question in SQL
Finance wants revenue per month and a running total. A common table expression builds the monthly figures, and a window function adds the running total on top.
import csv
import sqlite3
conn = sqlite3.connect("shop.db")
rows = conn.execute(
"""
WITH monthly AS (
SELECT substr(order_date, 1, 7) AS month,
ROUND(SUM(quantity * unit_price), 2) AS revenue
FROM orders
GROUP BY month
)
SELECT month,
revenue,
ROUND(SUM(revenue) OVER (ORDER BY month), 2) AS running_total
FROM monthly
ORDER BY month
"""
).fetchall()
with open("monthly_revenue.csv", "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["month", "revenue", "running_total"])
writer.writerows(rows)
for month, revenue, running in rows:
print(f"{month} {revenue:>9.2f} {running:>9.2f}")
print()
top = conn.execute(
"SELECT product, SUM(quantity) AS units FROM orders GROUP BY product ORDER BY units DESC"
).fetchall()
for product, units in top:
print(f"{product:<10} {units}")
Save it as report.py. SUM(revenue) OVER (ORDER BY month) is the key line: it adds up every month up to and including the current one. With the seed above the output is:
2025-01 5200.80 5200.80
2025-02 3669.00 8869.80
2025-03 2766.00 11635.80
2025-04 3885.00 15520.80
pen 254
backpack 233
lamp 219
notebook 194
Step 4: Run it as one job
Run the three scripts in order. Because each starts from a known state (DROP TABLE IF EXISTS in the load step, files overwritten in the others), you can run the whole sequence again at any time and get the same result. That property is called idempotence, and it is what makes a pipeline safe to re-run after a failure.
Check your work
SELECT COUNT(*) FROM ordersshould equal the number of loaded rows (300), not the number of lines in the CSV (302).- The last running total must equal
SELECT ROUND(SUM(quantity * unit_price), 2) FROM orders. - Two rejected rows is the right answer here. If you see zero, the bad rows were never added.
Where this stops scaling
SQLite is excellent for a single writer and data that fits on one machine. When several jobs must write at once, or the data outgrows a disk, the same SQL moves to a server database or a warehouse with very few changes. The habits stay the same: validate at the door, load in a transaction, and keep the job re-runnable.
Expected result
Two files appear, orders.csv and monthly_revenue.csv, and the last script prints revenue and a running total for each month plus units sold per product.
Troubleshooting
| Problem | Likely cause | Fix |
|---|---|---|
| sqlite3.OperationalError: near "(": syntax error, pointing at OVER | The SQLite library behind Python is older than 3.25, which added window functions. | Check sqlite3.sqlite_version. Upgrade Python, or replace the running total with a self-join. |
| sqlite3.IntegrityError: UNIQUE constraint failed: orders.order_id | You ran the load script twice without dropping the table. | The script drops and recreates the table first. If you removed that line, delete shop.db and run it again. |
Frequently asked questions
Why SQLite and not a bigger database?
It needs no server, lives in one file and speaks real SQL, so everything you learn here carries over to PostgreSQL, MySQL or a warehouse. Its limits are covered at the end.
Why reject bad rows instead of fixing them?
A pipeline should never silently guess. Rejecting and reporting keeps the data honest and gives you a list to take back to whoever produced the export.
Next steps
- Add a customers table and join it to orders
- Write the report to a spreadsheet-friendly file on a schedule
- Move the same SQL to a warehouse once the data no longer fits on one machine
References
From Grokem: https://www.grokem.com/data/data-analysis/python-sql/python-sqlite-analytics-pipeline
