First Python data shift

Do this after the Postgres lessons (through lesson 14 is ideal; lesson 16 is optional).

You are pretending to seed a tiny retail warehouse: generate realistic-ish random orders in Python, land them in Postgres, and verify counts. The website still runs nothing. You run every command under ~/dcubes/python-lab/shift. Docker must be running.

What you need

• Python venv skills from Virtualenvs and pip
• Compose Postgres pattern from Postgres from Python
• Docker engine up (same as the Docker module)

1. Workspace and venv

mkdir -p ~/dcubes/python-lab/shift/data
cd ~/dcubes/python-lab/shift
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install "psycopg[binary]"

cat > requirements.txt << 'EOF'
psycopg[binary]==3.2.3
EOF

If that exact psycopg version fails to resolve, run python -m pip install "psycopg[binary]" and python -m pip freeze | grep -i psycopg to pin what you actually got.

2. Start Postgres

cat > compose.yaml << 'EOF'
services:
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: dcubes
      POSTGRES_PASSWORD: dcubes
      POSTGRES_DB: retail
    ports:
      - "15432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U dcubes -d retail"]
      interval: 3s
      timeout: 3s
      retries: 20

volumes:
  pgdata:
EOF
docker compose up -d
docker compose ps

Wait until the service is healthy. Host port is 15432 (not 5432).

3. Generate random retail orders

This script uses only the standard library (random, csv, datetime). A fixed seed makes re-runs comparable.

cat > generate_orders.py << 'EOF'
#!/usr/bin/env python3
"""Generate random retail order lines and write data/orders.csv."""

from __future__ import annotations

import argparse
import csv
import random
from datetime import datetime, timedelta, timezone
from pathlib import Path

STORES = [f"STR-{i:02d}" for i in range(1, 6)]
PAYMENT_METHODS = ["card", "cash", "wallet"]
STATUSES = ["completed", "completed", "completed", "completed", "returned", "cancelled"]

CATALOG = [
    ("SKU-100", "Whole milk 1L", "grocery", 1.29),
    ("SKU-101", "Sourdough loaf", "grocery", 3.49),
    ("SKU-102", "Free-range eggs 12", "grocery", 4.15),
    ("SKU-110", "USB-C cable", "electronics", 9.99),
    ("SKU-111", "AA batteries 8pk", "electronics", 6.50),
    ("SKU-112", "Wireless mouse", "electronics", 18.00),
    ("SKU-120", "Dish soap", "home", 2.75),
    ("SKU-121", "Trash bags 30ct", "home", 5.20),
    ("SKU-122", "LED bulb 2pk", "home", 7.80),
    ("SKU-130", "Notebook A5", "office", 2.10),
    ("SKU-131", "Ballpoint pens 10", "office", 3.40),
    ("SKU-132", "Masking tape", "office", 1.95),
]


def random_order_ts(rng: random.Random, days: int = 60) -> datetime:
    now = datetime.now(timezone.utc)
    delta = timedelta(
        days=rng.randint(0, days),
        hours=rng.randint(0, 23),
        minutes=rng.randint(0, 59),
        seconds=rng.randint(0, 59),
    )
    return now - delta


def generate_rows(n: int, seed: int) -> list[dict[str, object]]:
    rng = random.Random(seed)
    rows: list[dict[str, object]] = []
    for i in range(1, n + 1):
        sku, name, category, unit_price = rng.choice(CATALOG)
        quantity = rng.randint(1, 5)
        line_total = round(quantity * unit_price, 2)
        customer = f"CUST-{rng.randint(1, 80):04d}"
        rows.append(
            {
                "order_id": f"ORD-{i:06d}",
                "order_ts": random_order_ts(rng).isoformat(),
                "store_id": rng.choice(STORES),
                "customer_id": customer,
                "sku": sku,
                "product_name": name,
                "category": category,
                "quantity": quantity,
                "unit_price": unit_price,
                "line_total": line_total,
                "payment_method": rng.choice(PAYMENT_METHODS),
                "status": rng.choice(STATUSES),
            }
        )
    return rows


def main() -> None:
    parser = argparse.ArgumentParser(description="Generate retail orders CSV")
    parser.add_argument("-n", type=int, default=300, help="number of order lines")
    parser.add_argument("--seed", type=int, default=42, help="RNG seed")
    parser.add_argument(
        "-o",
        type=Path,
        default=Path("data/orders.csv"),
        help="output CSV path",
    )
    args = parser.parse_args()

    rows = generate_rows(args.n, args.seed)
    args.o.parent.mkdir(parents=True, exist_ok=True)
    fieldnames = list(rows[0].keys())
    with args.o.open("w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)

    print(f"wrote {len(rows)} rows -> {args.o}")
    print("sample:", rows[0]["order_id"], rows[0]["sku"], rows[0]["line_total"])


if __name__ == "__main__":
    main()
EOF
python generate_orders.py -n 300 --seed 42

wrote 300 rows -> data/orders.csv
sample: ORD-000001 ...

Peek:

head -n 4 data/orders.csv
wc -l data/orders.csv

You should see a header plus 300 data lines (wc -l reports 301).

4. Load into PostgreSQL

cat > load_orders.py << 'EOF'
#!/usr/bin/env python3
"""Create orders table and load data/orders.csv with psycopg."""

from __future__ import annotations

import csv
import os
import sys
from pathlib import Path

import psycopg

CSV_PATH = Path("data/orders.csv")

CONN = dict(
    host=os.environ.get("PGHOST", "127.0.0.1"),
    port=int(os.environ.get("PGPORT", "15432")),
    dbname=os.environ.get("PGDATABASE", "retail"),
    user=os.environ.get("PGUSER", "dcubes"),
    password=os.environ.get("PGPASSWORD", "dcubes"),
)

DDL = """
CREATE TABLE IF NOT EXISTS orders (
  order_id TEXT PRIMARY KEY,
  order_ts TIMESTAMPTZ NOT NULL,
  store_id TEXT NOT NULL,
  customer_id TEXT NOT NULL,
  sku TEXT NOT NULL,
  product_name TEXT NOT NULL,
  category TEXT NOT NULL,
  quantity INTEGER NOT NULL,
  unit_price NUMERIC(10, 2) NOT NULL,
  line_total NUMERIC(12, 2) NOT NULL,
  payment_method TEXT NOT NULL,
  status TEXT NOT NULL
)
"""

INSERT = """
INSERT INTO orders (
  order_id, order_ts, store_id, customer_id, sku, product_name, category,
  quantity, unit_price, line_total, payment_method, status
) VALUES (
  %s, %s, %s, %s, %s, %s, %s,
  %s, %s, %s, %s, %s
)
"""


def read_rows(path: Path) -> list[tuple]:
    with path.open(newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        rows = []
        for r in reader:
            rows.append(
                (
                    r["order_id"],
                    r["order_ts"],
                    r["store_id"],
                    r["customer_id"],
                    r["sku"],
                    r["product_name"],
                    r["category"],
                    int(r["quantity"]),
                    r["unit_price"],
                    r["line_total"],
                    r["payment_method"],
                    r["status"],
                )
            )
        return rows


def main() -> None:
    if not CSV_PATH.is_file():
        print(f"missing {CSV_PATH} — run generate_orders.py first", file=sys.stderr)
        sys.exit(1)

    rows = read_rows(CSV_PATH)
    with psycopg.connect(**CONN) as conn:
        with conn.cursor() as cur:
            cur.execute(DDL)
            cur.execute("TRUNCATE orders")
            cur.executemany(INSERT, rows)
            cur.execute("SELECT count(*) FROM orders")
            total = cur.fetchone()[0]
            print("loaded_rows", total)

            cur.execute(
                """
                SELECT store_id, count(*) AS n
                FROM orders
                GROUP BY store_id
                ORDER BY store_id
                """
            )
            print("by_store")
            for store_id, n in cur.fetchall():
                print(f"  {store_id} {n}")

            cur.execute(
                """
                SELECT status, count(*) AS n
                FROM orders
                GROUP BY status
                ORDER BY n DESC, status
                """
            )
            print("by_status")
            for status, n in cur.fetchall():
                print(f"  {status} {n}")

            cur.execute(
                """
                SELECT category, round(sum(line_total), 2) AS revenue
                FROM orders
                WHERE status = 'completed'
                GROUP BY category
                ORDER BY revenue DESC
                """
            )
            print("completed_revenue_by_category")
            for category, revenue in cur.fetchall():
                print(f"  {category} {revenue}")
        conn.commit()


if __name__ == "__main__":
    main()
EOF
python load_orders.py

loaded_rows 300
by_store
  STR-01 ...
  ...
by_status
  completed ...
  ...
completed_revenue_by_category
  ...

Exact store and status counts depend on the seed and catalog weights; with --seed 42 and -n 300 they should be stable across re-runs on your machine.

5. Optional check with psql

If you still have the Postgres client habits from Docker:

docker compose exec db psql -U dcubes -d retail -c "SELECT count(*) FROM orders;"

 count
-------
   300

6. Re-run cleanly

Generate and load again anytime:

python generate_orders.py -n 300 --seed 42
python load_orders.py

TRUNCATE makes the load idempotent for this lab.

7. Tear down

docker compose down

Add -v only if you want to delete the Postgres volume.

Common mistakes

• Port 5432 — this lab publishes 15432.
• Database name lab vs retail — compose sets POSTGRES_DB: retail; match PGDATABASE.
• Loading before generate — data/orders.csv must exist.
• f-string INSERT — the loader uses parameterized %s values only.
• Host db from host Python — use 127.0.0.1.

What you practiced

• stdlib random data generation with a seed
• CSV as an intermediate contract
• psycopg DDL + executemany load
• verification queries (counts and simple aggregates)

Next

Next: the SQL module — query retail data with joins, CTEs, and windows. Then Git. Still guidance only, still on your machine.