Week 1 · Ingest anything › Day 2

Day 2. Engineering setup that survives a real job

1. Why this matters

Most "pipeline broke" incidents aren't clever bugs. They're a laptop-only dependency, a password in code, a path that only exists on one machine, or a log that says Error and nothing else. Five-year engineers set these up on day one, so you will too.

2. A project layout you'll reuse every week

retail-ingest/
├── pyproject.toml        # dependencies + tool config (one file)
├── uv.lock               # exact versions, committed
├── .env.example          # names of settings, no values
├── .gitignore            # .env, data/, .venv/
├── README.md
├── src/retail_ingest/
│   ├── __init__.py
│   ├── config.py         # reads settings once, typed
│   ├── log.py            # structured logging
│   └── sources/          # one module per source
├── tests/
└── data/                 # git-ignored: raw/, staging/

uv (from Astral) creates the venv, resolves dependencies and writes a lock file in seconds. pip + requirements.txt still works; the lock file is the part that matters.

pip install uv           # or see docs.astral.sh/uv for the installer
uv init --package retail-ingest && cd retail-ingest
uv add pandas polars duckdb openpyxl pydantic-settings python-dotenv
uv add --dev pytest ruff
uv run pytest

3. Config and secrets

Rule: code reads settings; settings live outside code. Locally that's a git-ignored .env. In production it's environment variables injected by the orchestrator or a vault (Azure Key Vault, Week 4).

# src/retail_ingest/config.py
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="P2D_")
    data_dir: Path = Path("./data")
    landing_dir: Path = Path("./data/landing")
    imap_host: str = "imap.example.com"
    imap_user: str = ""
    imap_password: str = ""          # never logged, never committed

settings = Settings()

4. Logs you can debug from

A good log line answers: what step, on which input, what happened, how many rows, how long. Use key=value or JSON so you can grep and parse it.

# src/retail_ingest/log.py
import json, logging, sys, time

class JsonFormatter(logging.Formatter):
    def format(self, r):
        base = {"ts": time.strftime("%Y-%m-%dT%H:%M:%S"), "level": r.levelname, "msg": r.getMessage()}
        base.update(getattr(r, "ctx", {}))
        return json.dumps(base)

def get_logger(name="p2d"):
    log = logging.getLogger(name)
    if not log.handlers:
        h = logging.StreamHandler(sys.stdout); h.setFormatter(JsonFormatter())
        log.addHandler(h); log.setLevel(logging.INFO)
    return log

# usage
log = get_logger()
log.info("file landed", extra={"ctx": {"source": "email", "file": "sales_R104.csv", "rows": 812, "sha256": "ab12…"}})

5. Claude Code, safely, from day two

You'll use Claude Code throughout the course as a pair programmer. Copy CLAUDE.md from your bi-skills-kit/claude-code/ into each project and trim it. The rules that matter most on a personal project: never print secrets, propose before applying on anything shared, and save evidence (queries and outputs) for every claim. Ask it to explain code it writes, and reject anything you can't explain back.