Spine Milestone 2: CityOps Intake Pipeline

Eight posts ago you were reading a CSV somebody handed you. Today the pipeline reads itself: it pulls the city's 311 feed every morning, checks every record against a contract it can actually run, quarantines what fails, stores what passes, and proves — with arithmetic, not vibes — that every record is accounted for. This is the milestone the whole stage was building toward.

Assumes: Posts 1–8. Installs: pip install requests pydantic pandas. Everything below ran against the real 311 open-data feed; every number shown is computed, not asserted.

Before you code: clarify the ask

Input: the city's 311 open-data feed, every morning
Output: a SQLite database Maria's reports read from, plus a reconciliation report: fetched vs stored vs quarantined, all adding up
Deadline: running unattended by next Monday

Three requirements fell out of that conversation, and they're the spec: reconciliation (every record accounted for), quarantine on drift (reject what you don't understand), idempotent reruns (running twice changes nothing).

The minimal concept

A data pipeline has five stages, and each one earns Maria's trust differently:

Stage	Does	Trust mechanism
Extract	Pull the feed (Post 6: sessions, timeouts)	Half-open date window — no record counted twice, none skipped
Validate	Check every record against the contract	The contract is runnable: schema drift is rejected by code, not by hope
Clean	Normalize (Post 4: strip, case, UTC instants)	Cosmetic only — cleaning never rescues a record the contract rejected
Load	Insert into SQLite (Post 3)	Dedupe key as PRIMARY KEY — reruns are no-ops by construction
Reconcile	Count everything, hash everything	fetched == stored + quarantined + duplicates, and the hashes match

Four ideas carry the milestone:

The data contract is a runnable artifact. A schema document in a wiki is a wish. A Pydantic model in the pipeline is a bouncer: every record is validated before it touches the database, and anything the model doesn't understand goes to the dead-letter queue. When the vendor renames a field overnight, you find out from the quarantine count, not from a broken dashboard three weeks later.

The DLQ is not a trash can. Quarantined records are stored with their raw payload, the validation error, and a timestamp. Nothing is silently dropped — "we threw away 16 rows" is a reportable fact with evidence attached, which is exactly what Maria asked for.

Idempotency is a design choice, not a retry policy. The dedupe key is the PRIMARY KEY. Inserting the same record twice is a no-op at the database level, so a failed 6 AM run can simply be rerun at 6:20. You don't need distributed locks; you need a key.

The audit log is the pipeline's memory. Every run writes one row: window, counts, hashes, timestamp. When Maria asks "what happened Tuesday?", the answer is a query, not a archaeology expedition.

Build it, part 1: extract and the v1 contract

Extraction is Post 6's pattern with a half-open window — [Sep 20, Sep 23), so a record stamped 23:59:59.500 can't fall through the cracks (Post 5's boundary lesson, applied to queries):

def fetch_311(session, start, end, page_size=1000):
    """Paginated extract with stable ordering. A 3-day window can hold
    far more than one page — without pagination, "complete" is a lie."""
    rows, offset = [], 0
    while True:
        resp = session.get(
            "https://data.cityofnewyork.us/resource/erm2-nwe9.json",
            params={
                "$select": "unique_key,created_date,agency,complaint_type,"
                           "descriptor,borough,incident_zip,city",
                "$where": f"created_date >= '{start}T00:00:00' "
                          f"AND created_date < '{end}T00:00:00'",
                # Stable sort: without $order, page boundaries are arbitrary
                "$order": "unique_key",
                "$limit": str(page_size),
                "$offset": str(offset),
            },
            timeout=(5, 30),
        )
        resp.raise_for_status()
        page = resp.json()
        rows.extend(page)
        if len(page) < page_size:
            break
        offset += page_size
    return rows

Two honesty notes on pagination. First, offset pagination against a live feed isn't a perfect snapshot — records inserted mid-pull can shift page boundaries. For a reporting window that's already closed (yesterday's data, pulled this morning), the feed is effectively static and this is fine. For a live-updating feed, you'd need a cursor or a snapshot API. Second, verify completeness independently: fetch the source count ($select=count(*) with the same $where) and confirm len(rows) == source_count before reconciling.

Then the contract — v1, the straightforward version. One critical definition first: the 311 feed's created_date values are NYC wall-clock time without an offset (e.g. 2026-09-20T08:00:00.000). Post 5's parse_ts rejects naive timestamps — and rightly so. So the contract must explicitly localize: attach America/New_York (the feed's documented timezone), handle the ambiguous/nonexistent local times around DST transitions, then convert to UTC. Never silently attach UTC to a naive timestamp — 8 AM in New York is not 8 AM UTC:

from zoneinfo import ZoneInfo

NYC = ZoneInfo("America/New_York")

def localize_nyc_wallclock(v: str) -> datetime:
    """'2026-09-20T08:00:00.000' -> UTC instant. The feed documents
    Eastern wall time; we make that assumption explicit and testable."""
    naive = datetime.fromisoformat(v)  # no offset in the source
    # fold=0/1 disambiguates the repeated 1 AM hour on fall-back;
    # nonexistent spring-forward times raise, and should — loudly.
    return naive.replace(tzinfo=NYC).astimezone(timezone.utc)

The validator uses this, not parse_ts directly — different source semantics, different boundary function, same UTC discipline:

from pydantic import BaseModel, ConfigDict, Field, field_validator
from typing import Optional

class ServiceRequestV1(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    unique_key: str
    created_at: datetime = Field(alias="created_date")  # feed name != our name
    agency: str
    complaint_type: str
    descriptor: Optional[str] = None
    borough: Optional[str] = None
    incident_zip: str = Field(min_length=5, max_length=5)
    city: Optional[str] = None

    @field_validator("created_at", mode="before")
    @classmethod
    def _ts(cls, v):
        try:
            return localize_nyc_wallclock(v)  # NYC wall -> UTC instant; garbage raises
        except (ValueError, TypeError) as e:
            raise ValueError(f"bad timestamp {v!r}") from e

The validation gate — Post 2's quarantine instinct, now as a function. Note what it catches: only to add context (the DLQ entry) — the error itself propagates into the quarantine record:

def validate(rows, model):
    """Contract gate: valid models out, DLQ entries for the rest."""
    valid, dlq = [], []
    for raw in rows:
        try:
            valid.append(model(**raw))
        except ValidationError as e:
            dlq.append({"raw": raw, "error": e.errors()[0]["msg"]})
    return valid, dlq

Cleaning is Post 4's toolkit doing what it's for — cosmetic normalization, nothing structural:

def clean(valid):
    df = pd.DataFrame([r.model_dump() for r in valid])
    for col in ["agency", "complaint_type", "descriptor", "borough", "city"]:
        df[col] = df[col].str.strip()
    df["borough"] = df["borough"].str.upper()
    df["created_at"] = pd.to_datetime(df["created_at"], utc=True)
    # ...back to models, NaN -> None (the contract speaks None, not NaN)

Loading is Post 3's SQL with the dedupe key as PRIMARY KEY. v1 dedupes on unique_key — the feed's own id, the obvious choice:

CREATE TABLE service_requests(
  fingerprint TEXT PRIMARY KEY,   -- v1: this holds unique_key; v2: content hash
  unique_key TEXT NOT NULL,
  created_at TEXT NOT NULL,        -- UTC ISO instant
  agency TEXT NOT NULL,
  complaint_type TEXT NOT NULL,
  descriptor TEXT, borough TEXT, incident_zip TEXT, city TEXT,
  ingested_at TEXT NOT NULL
);
-- INSERT OR IGNORE: the PRIMARY KEY makes reruns no-ops by construction

First run, v1, against the real feed — 1000 rows in, plus 5 deliberately poisoned rows (injected to demonstrate the DLQ without waiting for the feed to misbehave):

fetched: 1005 | valid: 989 | quarantined: 16 | stored: 989
adds_up: true   (989 + 16 = 1005)
hash_match: true

The arithmetic holds. But look at the DLQ: 16 quarantined, and only 5 were my poison. Eleven real rows failed the contract. Before you read on — that's the pipeline working correctly and the contract being wrong. The question is why.

Build it, part 2: the v2 contract

The updated model. incident_zip becomes Optional with a normalizing validator — a 5-digit zip is kept, anything else becomes NULL instead of failing the row. The row survives; only the unreliable value is quarantined, in place:

class ServiceRequest(BaseModel):
    # ... same fields, except:
    incident_zip: Optional[str] = None  # unreliable: normalize, never reject

    @field_validator("incident_zip", mode="before")
    @classmethod
    def _zip(cls, v):
        if v is None:
            return None
        z = str(v).strip()
        return z if (len(z) == 5 and z.isdigit()) else None

And identity gets a redesign — but not the one the first draft tried. The first instinct was to replace unique_key with a content fingerprint. That's wrong, and worth understanding why: two residents can file separate complaints with identical timestamps, agency, type, and borough. A content hash merges genuinely different requests — the run found 10 such pairs — and for a government reporting system, silently undercounting real complaints is a data-integrity failure, not a tradeoff.

The correct design keeps unique_key as the primary identity — it's the source's own identifier, and you don't override the source's identity without verified evidence. Instead, the pipeline distinguishes three cases:

• Exact duplicate: same unique_key already in the database. INSERT OR IGNORE handles it — a true no-op.
• Possible duplicate: new unique_key, but the content fingerprint matches an existing row. Both records are preserved; the new one is flagged possible_duplicate_of for investigation, not merged.
• Updated record: same unique_key with changed fields. Stored as a new version, linked to the original — corrections are visible, not silently absorbed.

The fingerprint is still computed — as a suspicion signal, not an identity. When identity is uncertain, preserve evidence rather than silently merging data:

def content_fingerprint(self) -> str:
    """Suspicion signal only — NEVER used as the dedupe key."""
    core = "|".join([
        self.created_at.isoformat(), self.agency, self.complaint_type,
        self.descriptor or "", self.borough or "", self.city or "",
    ])
    return hashlib.sha256(core.encode()).hexdigest()

CREATE TABLE service_requests(
  unique_key TEXT PRIMARY KEY,   -- the source's identity: trusted, not replaced
  created_at TEXT NOT NULL,       -- UTC ISO instant
  agency TEXT NOT NULL,
  complaint_type TEXT NOT NULL,
  descriptor TEXT, borough TEXT, incident_zip TEXT, city TEXT,
  content_fingerprint TEXT NOT NULL,
  possible_duplicate_of TEXT,     -- NULL normally; set when fingerprint matches
  version INT NOT NULL DEFAULT 1,  -- increments on same-key updates
  ingested_at TEXT NOT NULL
);

Same intake, v2 contract — with explicit per-run naming, because after reruns the database holds cumulative state and the report must describe this run:

fetched_this_run: 1005
  valid_this_run: 1000  (= inserted_this_run 990 + already_present 0 + flagged_possible_dupes 10)
  quarantined_this_run: 5
adds_up: true   (1000 + 5 = 1005)
source_hash_match: true    # raw payloads match the feed's bytes
stored_hash_match: true    # accepted records match what's in the table

The 11 zip-less rows are back in — incident_zip NULL, row intact. The DLQ holds exactly the 5 poison rows. The 10 same-content pairs are preserved and flagged, not merged — each keeps its own unique_key, with possible_duplicate_of pointing at its twin for a human to adjudicate. Every record accounted for; no real-world event silently uncounted.

Two separate hash checks, because they prove different things. Source integrity: hash the raw JSON bytes as received — this proves the extract didn't corrupt or truncate anything in transit. Stored integrity: hash the canonical form of the accepted records (sorted keys, fixed field order) and compare against the same canonicalization of the table contents — this proves the load wrote exactly what validation accepted. Neither proves business completeness (the feed could be missing records upstream); they prove the pipeline didn't lose or alter what it was given.

Break it, three ways

A pipeline that only works on good days is a demo. Here's what the bad days look like — all three reproduced against the real feed.

1. The poison rows. The 5 injected rows landed in the DLQ with their raw payloads and real validation errors — this is what "quarantine with evidence" means:

POISON2 -> Value error, bad timestamp 'yesterday-ish'
POISON3 -> Input should be a valid string        # agency: None
POISON4 -> Value error, bad timestamp {'epoch': 1726870000}
POISON5 -> Field required                        # complaint_type missing

Each DLQ row stores the raw JSON, the error, and the timestamp. Maria never has to take "we dropped some rows" on faith — she can read exactly which rows and why.

2. Schema drift. The vendor renames complaint_type to complainttype overnight — no announcement, just a quiet Tuesday. The contract is a bouncer, not a janitor:

fetched: 1000 | valid: 0 | quarantined: 1000 | stored: 0
adds_up: true
rows in service_requests after the drifted batch: 0

Every single row rejected, zero rows polluting the table, and the reconciliation still adds up. The 6 AM alert isn't "the dashboard is wrong" — it's "quarantine spiked to 1000, the feed changed shape." That's the failure mode you want: loud, contained, and reversible. You fix the contract (or call the vendor), rerun, and the PRIMARY KEY dedupe means the rerun only fills the gap.

3. The re-emissions (simulated). Three real rows, re-emitted under new unique_key values with "corrected" zips — the scenario Dev's team suspects. Under v1's key-based dedupe, all three store as new rows: 992 instead of 989, silently inflating Maria's counts. Under v2: the unique_key PRIMARY KEY accepts them (they're new keys), but the content fingerprint matches existing rows — so all three are stored and flagged as possible_duplicate_of their twins. Nothing is merged, nothing is lost, and a human sees exactly which pairs need adjudication. The double-count is visible; the evidence is preserved.

Productionize: the 6 AM machine

The pipeline now needs to run itself. Three pieces, all from the last two posts' toolkit:

Schedule. One cron line — the pipeline already takes --start/--end, so "yesterday" is just arguments:

30 6 * * * /opt/cityops/venv/bin/python /opt/cityops/intake.py \
  --start yesterday --end today >> /var/log/cityops/intake.log 2>&1

CLI + logging. argparse for the window and limit, structured logging with the run_id on every line — so when the 6:20 rerun happens, its log lines don't interleave ambiguously with the 6:00 attempt's. (This is Post 8's packaging-and-logging harness, with the pipeline living inside it.)

The test suite. Post 7's discipline, and it replays the saved feed payload — no network in unit tests:

test_contract_accepts_good_row ............ ok
test_contract_quarantines_poison .......... ok
test_reconciliation_adds_up ............... ok
test_rerun_is_idempotent .................. ok
test_reemission_caught_by_fingerprint ..... ok

5 passed

And the audit log — one row per run, so "what happened Tuesday?" is a query:

CREATE TABLE audit_log(
  run_id TEXT PRIMARY KEY, window_start TEXT, window_end TEXT,
  fetched INT, valid INT, quarantined INT, duplicates INT, stored INT,
  source_hash TEXT, stored_hash TEXT, hash_match INT, finished_at TEXT
);

Explain it to the customer

"Maria — this morning's pull: 1,005 records in, 990 newly stored, 5 quarantined, 10 flagged as possible duplicates. The 5 quarantined are malformed feed rows — they're in the DLQ table with the exact errors, nothing was dropped silently. The 10 flagged pairs have matching content under different source IDs; both records are preserved, and the flags are waiting for your team's adjudication — we don't merge what we can't prove is the same event. Both hash checks pass: the raw bytes match the feed, and the table matches what validation accepted. If anything looks off, rerunning is safe — the source-key PRIMARY KEY makes reruns no-ops. And one thing the pipeline taught us: the feed's ZIP codes can't be trusted, so they're stored but never used for identity."

The pattern, one last time: the numbers, then the caveats, then the rerunnable method. Maria doesn't read the pipeline — she reads the proof the pipeline produces.

Where this lands in CityOps

The intake pipeline is the second spine of CityOps. Milestone 1 gave the project its skeleton; Milestone 2 gives it a heartbeat — a database that refills itself every morning, with proof. Everything downstream reads from service_requests now: the weekly report, the weather enrichment, and in Stage 3, the API itself. The contract file is the most important artifact you produced this stage: it's the executable version of "what we believe about the feed," and every future change to the feed starts by changing it.

And the stage exit check, stated plainly: you can now take "clean up this data and automate it" from a vague request to a tested, documented, scheduled, reconciled pipeline. That's the whole of Stage 2 in one sentence — and it's the job.

Post 4's principle was if you can't rerun it, you didn't clean it. Post 6's was don't trust the happy path. Milestone 2's: every record accounted for — fetched, stored, quarantined, or flagged, the arithmetic always closes. And the deeper lesson: accounted for doesn't mean merged away. When identity is uncertain, preserve the evidence and let a human decide.