Warehouse Data Pipeline

This data pipeline is a full rework of an old data engineering assignment I did a couple of years ago. The scenario is that large store chains that receive messy TSV files from their wholesale suppliers, which need to be cleaned and adjusted for their Warehouse Management Systems (WMS).

The pipeline cleans and corrects the data into a WMS-acceptable format. The old assignment defined a couple of its own business rules/logic, and a couple of suggestion where you, the developer, needed to make decisions in.

For this rewrite, I renamed this list of business rules/logic/assumptions the WMS Contract. The pipeline will clean, drop, and quarantine rows from the input data based on the WMS contract.

This project was an excuse for me to use/try out Polars' LazyFrames, which is optimized for large datasets. It's also using FastAPI for providing results, Pydantic's Settings implementation for configuration. I also created a script for generating synthetic data, so the pipeline can be easily used locally.

How to run

Prerequisites

• Python >=3.13
• Docker or an OCI-compliant equivalent
• An interest in data pipelines

Environment variables

The pipeline needs an ENCRYPTION_KEY (32 random bytes, url-safe base64) for the reversible SKU generator. Start with the example file:

cp src/warehouse_pipeline/.env.example src/warehouse_pipeline/.env

Then generate a key. The snippet imports cryptography, so it must run from an environment that has the deps installed:

# with uv (no manual install needed):
uv run python -c "import base64; from cryptography.hazmat.primitives.ciphers.aead import AESSIV; print(base64.urlsafe_b64encode(AESSIV.generate_key(bit_length=256)).decode())"

# with pip (run AFTER `pip install -r requirements.txt` from the section below):
python -c "import base64; from cryptography.hazmat.primitives.ciphers.aead import AESSIV; print(base64.urlsafe_b64encode(AESSIV.generate_key(bit_length=256)).decode())"

Paste the output into .env after ENCRYPTION_KEY=.

with uv

uv sync                                                       # install deps

uv run python src/warehouse_pipeline/main.py                  # one-shot pipeline run
uv run uvicorn src.warehouse_pipeline.api.api:app --reload    # API (runs the pipeline on startup)

uv run pytest tests/                                          # tests

Shortcuts supported by the Makefile: make run-pipeline, make serve, make test.

with pip

python3 -m venv venv && source venv/bin/activate
pip install -r requirements.txt
export PYTHONPATH=src

python3 src/warehouse_pipeline/main.py
uvicorn src.warehouse_pipeline.api.api:app --reload

pytest tests/

with Docker

The image doesn't embed a key; pass one at run time:

docker build -t warehouse-data-pipeline .

# Generate a key on the host (`uv` shown; swap for plain `python` if you've
# installed the deps via pip):
KEY=$(uv run python -c "import base64; from cryptography.hazmat.primitives.ciphers.aead import AESSIV; print(base64.urlsafe_b64encode(AESSIV.generate_key(bit_length=256)).decode())")
docker run -d -p 8000:8000 -e ENCRYPTION_KEY="$KEY" warehouse-data-pipeline

API usage with curl

# Cleaned WMS load (default format CSV)
curl "http://localhost:8000/wms_data"
curl "http://localhost:8000/wms_data?format=tsv"
curl "http://localhost:8000/wms_data?format=json"

# Rows the pipeline excluded for data-quality reasons (each row has a 'reason' column)
curl "http://localhost:8000/wms_quarantine?format=csv"

Every file response carries an X-Pipeline-Duration header with the last pipeline run's seconds.

Synthetic input data (aside)

The TSV files in data/input/ come from a seeded generator at scripts/generate_synthetic_data.py. Running it overwrites the inputs with the same deterministic dataset (14,000 articles across six European languages, plus all the data-quality quirks the pipeline is built to handle). Run only if you want to regenerate:

uv run python scripts/generate_synthetic_data.py

The WMS Contract

The WMS only ingests rows that satisfy this contract. A row that doesn't is either quarantined (collected in a separate file with a reason column for a data engineer to inspect) or silently excluded (correct business filtering, logged as a count). Two layers: hard rules inherited from the original assignment, and soft assumptions chosen here for the ambiguous bits of the data.

Hard rules (from the assignment)

Non-negotiable. A row that breaks any of these does not enter the WMS.

Rule	What it means	Where it's applied
In-assortment only	assortm_code == 1	Cleaner.clean_articles
Output schema	10 columns: own_sku, ean_list, content, content_unit, description, temperature_zone, shelf_life, volume_mm3, brutto_weight, brutto_weight_unit	Settings.EXPORT_COLUMNS, enforced in ContractGate
No nulls in the output	Every export field must carry a value	ContractGate
Per-zone volume limit	ambient ≤ 36,366 cm³ · chilled ≤ 29,092 cm³ · frozen ≤ 6,732 cm³. Strict > — exactly at the limit passes; one cm³ over is quarantined as exceeds_volume_limit.	Settings.VOLUME_LIMITATIONS, ContractGate._status_expr
own_sku reversible & multi-supplier	Same (supplier, article_id) → same own_sku across runs; the supplier name in the payload keeps SKUs disjoint across suppliers	SupplierSKUGenerator (AES-SIV)

Soft assumptions (chosen by me)

These are my reads of ambiguous bits of the data. The posture is conservative: if a guess would risk a hard rule, the row goes to quarantine instead.

Units are inferred per column

Neither input file carries unit metadata. Inferred from the value range:

• Article dimensions: centimetres. Values 1–50; sub-millimetre would be physically meaningless.
• Trade-unit dimensions: millimetres. Values min 90 – max 3,000 (median 275); as cm that's 0.9–30 m boxes (absurd for 6–20 retail items), as mm it's 9–300 cm (a realistic hand-carry case).

Applied in Processor._calculate_volumes via TRADE_UNIT_DIM_DIVISOR_TO_CM3 = 1000 (mm³ → cm³).

Volume comes only from real measurements (no mass → volume)

First-match-wins hierarchy in Processor._calculate_volumes:

1. Article dimensions l × d × h (cm).
2. Trade per-unit volume (l × d × h) / cu / 1000. Per-volume (not per-dimension) division avoids the ÷cu³ pitfall.
3. Volume-unit content × factor, only for ML (× 1) and LT (× 1000).

There is intentionally no mass-as-volume fallback. Density-1 for solids is off 2–20× for typical retail items (chips ≈ 0.05 g/cm³, flour ≈ 0.6, salt ≈ 1.2, chocolate ≈ 1.3) — the WMS contract cannot tolerate that. Mass-only rows without dims land in quarantine as underivable_volume.

Brutto weight: real measurements or a small packaging-overhead estimate

Three tiers in Processor._calculate_brutto_weight, first match wins. There is intentionally no fallback from a piece count — storing "12 eggs" as brutto_weight = 12 ST is not a weight, so ST/MT articles without a real weight are quarantined as underivable_brutto_weight.

1. Supplier's brutto_weight, kept as-is. Unit follows content_unit with ML→GR, LT→KG, ST→GR.
2. Trade-unit total weight (kg) ÷ consumer_units_in_trade_unit × 1000 → grams.
3. Content × BRUTTO_PACKAGING_OVERHEAD (default 1.05, i.e. content + 5% packaging mass — a lower-bound estimate of real retail packaging, which is 3–15%). For liquids this also leans on density-1, which is acceptable for brutto weight (±15% for typical retail liquids). Same unit substitutions as Tier 1.

Trade-unit dedup picks one whole row

For duplicate SKUs in the trade-units file, keep the one variant with the largest per-consumer-unit volume (tie-break: per-unit weight). Zero/null-measurement rows score 0 and only win when no real variant exists. Per-column maxima would synthesise a record that matches no real package and propagate internally-inconsistent values downstream.

Applied in Cleaner.clean_trade_units.

Temperature-zone repairs

• The Dutch value diepvries is normalised to frozen (the limits config is English).
• Product groups missing from the supplier's temperature_zone.tsv are recovered from a manually curated missing_temperature_zones.tsv.

Applied in Cleaner.clean_temperature_zones.

Shelf life is a fixed per-zone default

ambient 548 days (18 months), chilled 10 days, frozen 365 days. Defaults match typical retail-supply norms. Unknown zone → null → quarantined.

Applied in Processor._generate_shelf_life.

What happens to a row that breaks the contract

ContractGate is the gate. Every violation lands in the quarantine with one of these reasons (first match wins):

1. unknown_temperature_zone — zone is null or not in the limits config.
2. underivable_volume — no volume tier produced a positive value.
3. underivable_brutto_weight — no brutto-weight tier produced a positive value (typically an ST/MT article without a real weight).
4. null_in_export_column — any other null leaked through.
5. exceeds_volume_limit — passes every data-quality check, but its volume is over the per-zone WMS limit.

The list lives in Settings.QUARANTINE_REASONS, so reorder or extend in one place.

Pipeline structure

Six single-responsibility stages, orchestrated by WMSPipeline.run inside one try/except boundary. Polars LazyFrames flow through stages 1–4; collection happens in ContractGate.

1. Loader (loader.py) — reads each TSV (double-tab field separator), normalises trailing-space headers ("height " → "height"), selects schema columns and casts types. No null filling, no filtering, no business logic.

2. Cleaner (cleaner.py) — data-quality normalisation:

   • Articles: keep only assortm_code == 1 (in the WMS' assortment).
   • Trade units: a supplier SKU can appear under many packaging variants (a 6-pack, a 20-case, …). EANs aggregate over all variants; measurements come from one whole row — the variant with the largest per-consumer-unit volume (tie: per-unit weight). That's the worst-case packaging a WMS bay must hold.
   • Temperature zones: normalise the Dutch diepvries → frozen (the limits config is English) and append a manually curated mapping for product groups the supplier file omits.

3. Joiner (joiner.py) — left-joins everything onto articles; trade-unit measurement columns are carried under a _trade suffix where they collide. Aggregates trade + consumer EANs into one sorted comma-separated ean_list per SKU.

4. Processor (processor.py) — enriches: brutto weight + unit, volume (cm³ and mm³), shelf life, own_sku. Does not drop rows; the brutto-weight and volume tier fallbacks implement the soft assumptions from the contract.

5. ContractGate (contract_gate.py) — the contract gate. Partitions the processed rows into clean and quarantined (each violator tagged with a reason), per the WMS contract above.

6. Exporter (exporter.py) — writes the clean load as wms_load.{csv,tsv,json} and the quarantine as quarantine_<timestamp>.{csv,tsv,json} per run (timestamped files are gitignored). JSON splits ean_list from a comma-string into a real list.