Building a zoning change notification pipeline in Python

You wire a change-detection job to a webhook, ship it, and for a week it is perfect. Then on the first Sunday of the month Maricopa County re-publishes its entire zoning layer as part of a routine GIS refresh, your detector faithfully reports forty thousand “changes,” and every subscriber wakes up to hundreds of identical pages for parcels they have watched unchanged for years. A quieter version of the same bug is worse: an analyst never receives an alert for a genuine downzoning because a retry fired twice, the second delivery was deduplicated, and somewhere a race dropped the first. This guide builds the notifier that avoids both — one that watches change events and alerts only the right subscribers, exactly once, and stays silent when a county re-publishes a layer wholesale. It is the hands-on companion to the zoning change alerting topic and sits inside the broader Spatial Impact Analysis & Zoning Change Detection framework.

Diagnosis: tracing the duplicate storm and the silent miss

Before writing a line of the fix, reproduce both failure modes and read their signatures, because they demand opposite remedies. A duplicate storm and a missed alert can both look like “the alerting is broken,” but one is over-delivery and the other is under-delivery.

The duplicate storm has a clear fingerprint in the logs — the same parcel and subscriber pair, delivered many times in a tight window, all carrying an identical old-to-new zoning transition:

Four idempotency keys, and which failure each one actually prevents A four-row by four-column matrix of idempotency key designs against failure modes. Rows are: no key at all, a key derived from the run identifier, a key derived from the parcel identifier alone, and a key derived from the parcel identifier together with the canonical change hash. Columns are whether the key prevents duplicates on a retry, on a full reprocess, whether it still allows a genuine second change to the same parcel to be delivered, and whether it survives a subscriber replay. Only the composite key prevents both duplicate classes while still letting a real second change through. Four idempotency keys, and which failure each one actually prevents dupes on a retry dupes on a full reprocess second real change still delivered? survives a subscriber replay No idempotency key no no yes no key = run_id within a run no — new run id yes no key = parcel_id yes yes no — swallowed yes key = parcel_id + change hash yes yes yes yes correct partial fails
The third row is the trap: keying on parcel_id alone stops the duplicate storm and also swallows the next real rezone of that parcel, because it looks like the same event. The key has to identify the change, not the parcel — which is what pairing the identifier with a canonical change hash does.
delivered watcher=portfolio-7 parcel=301-44-118 R1->R1 sent_at=1720915200
delivered watcher=portfolio-7 parcel=301-44-118 R1->R1 sent_at=1720915201
delivered watcher=portfolio-7 parcel=301-44-118 R1->R1 sent_at=1720915203

Two tells here: R1->R1 is not a change at all (the classifier should have ranked it INFO and the detector arguably should not have emitted it), and the three sends are milliseconds apart. That pattern is a re-published layer flowing straight through with no deduplication and no reprocess suppression.

The silent miss is harder to see because its evidence is an absence. You confirm it by reconciling: take a change event you know occurred and grep the delivery ledger for its idempotency key. When the miss is a race, you find the key claimed and then an error, with no successful send:

claim key=alert:9f2c… ok=True
deliver watcher=study-area-3 parcel=118-06-042 attempt=0 error=ClientConnectorError
deliver watcher=study-area-3 parcel=118-06-042 attempt=1 error=ClientConnectorError
dead-letter watcher=study-area-3 parcel=118-06-042 reason=retries_exhausted

That is actually the correct behaviour — the event was dead-lettered, not lost. The genuine bug is the version where the key is claimed, delivery fails, and the key is never released or dead-lettered, so a retry of the whole batch finds the key already present and skips it forever. The pipeline below closes that gap by dead-lettering on exhaustion and never silently abandoning a claimed key.

Decision flow for the zoning change notification pipeline A change event enters and is first tested for whether it is part of a bulk reprocess. If yes, its idempotency key is primed into the store and no alert is sent. If no, the event is matched against the subscription index; unmatched events are recorded and dropped. Matched watcher pairs hit the idempotency gate: if the key is already claimed the event is counted as a duplicate and dropped, otherwise the key is claimed and delivery is attempted with retry. A successful delivery is written to the audit ledger; delivery that exhausts its retries is written to the dead-letter store, which is also audited. no matched yes: reprocess key claimed new sent retries spent Change event parcel · zoning Reprocess? suppression gate Match watchers STRtree · AOI Idem gate claim key Deliver + retry backoff · jitter Prime key no alert sent Duplicate dropped Audit ledger provable send Dead-letter store replayable · audited
Every event resolves to exactly one terminal state: primed, dropped as duplicate, delivered and audited, or dead-lettered — never silently lost.

Step-by-step implementation

Each step owns one concern. Assemble them into the runner in the final step.

Step 1 — Build the subscription index

A subscriber is an area of interest plus a delivery target and a minimum severity. Reproject every watcher polygon to the working CRS once, then build an STRtree so matching is a logarithmic-time envelope query rather than a linear scan. Keep the exact intersects test after the query, because the tree only filters by bounding box.

from dataclasses import dataclass
from shapely import STRtree
from shapely.geometry.base import BaseGeometry

@dataclass(frozen=True)
class Subscriber:
    watcher_id: str
    aoi: BaseGeometry     # already reprojected to the working CRS
    channel: str          # "webhook" | "queue"
    target: str
    min_severity: int = 2 # NOTICE and above

class SubscriberIndex:
    def __init__(self, subscribers):
        self._subs = list(subscribers)
        # STRtree is static; rebuild it when the registry changes, not per event
        self._tree = STRtree([s.aoi for s in self._subs])

    def watchers_for(self, parcel_geom: BaseGeometry):
        if parcel_geom.is_empty:
            return []
        candidates = self._tree.query(parcel_geom)  # indices of bbox overlaps
        return [self._subs[i] for i in candidates
                if self._subs[i].aoi.intersects(parcel_geom)]

Step 2 — Match the changed parcels and rank severity

Turn each incoming change event into (subscriber, event, severity) triples. Rank by land-use impact and parcel area, not by whatever label the feed attached, and drop the pair immediately when the severity falls below the subscriber’s threshold so low-value noise never reaches the delivery stage.

from shapely.geometry import shape

def severity_of(old_code: str, new_code: str, parcel_geom) -> int:
    if old_code == new_code:
        return 1  # INFO: not a real land-use change
    intensive = {"C3", "M1", "M2", "I1", "I2"}
    residential = {"R1", "R2", "R3"}
    upzoned = old_code in residential and new_code in intensive
    try:
        large = parcel_geom.area > 5_000.0
    except Exception:
        large = False
    if upzoned and large:
        return 4  # CRITICAL
    return 3 if upzoned else 2  # MATERIAL else NOTICE

def match_events(index: SubscriberIndex, events):
    for ev in events:
        geom = shape(ev["geometry"])
        sev = severity_of(ev["old_zoning"], ev["new_zoning"], geom)
        for sub in index.watchers_for(geom):
            if sev >= sub.min_severity:
                yield sub, ev, sev

Step 3 — Dedupe with an idempotency key and TTL

Compute a deterministic key over only the fields that give the alert meaning, and claim it atomically. The reprocess flag routes bulk re-publishes to prime, which records the key so a later genuine change still dedupes but sends nothing now. This is the single gate that prevents the duplicate storm.

import hashlib

def idem_key(watcher_id: str, ev: dict) -> str:
    parts = "|".join([watcher_id, ev["parcel_id"], ev["jurisdiction"],
                      ev["old_zoning"], ev["new_zoning"], ev["effective_date"]])
    return "alert:" + hashlib.sha256(parts.encode()).hexdigest()

class Deduper:
    def __init__(self, redis, ttl=7 * 24 * 3600):
        self._redis, self._ttl = redis, ttl

    async def is_new(self, key: str) -> bool:
        # SET NX EX is atomic: the loser of a race gets False and stops
        return bool(await self._redis.set(key, "1", nx=True, ex=self._ttl))

    async def prime(self, key: str) -> None:
        # record the key WITHOUT signalling a send (suppression path)
        await self._redis.set(key, "1", nx=True, ex=self._ttl)

Step 4 — Deliver with retry and a dead-letter fallback

Delivery is at-least-once. Retry transient failures with capped exponential backoff and jitter; on exhaustion, write the payload and reason to a dead-letter store rather than dropping it. Bounding concurrency with a semaphore keeps a metro-wide change from opening thousands of sockets at once.

import asyncio, random, logging
import aiohttp

logger = logging.getLogger("notifier")
MAX_RETRIES, BASE_DELAY = 4, 0.5

async def deliver(session, sub, body, dead_letter):
    for attempt in range(MAX_RETRIES):
        try:
            if sub.channel == "webhook":
                async with session.post(sub.target, json=body, timeout=15) as r:
                    r.raise_for_status()
            elif sub.channel == "queue":
                await session.app_queue.put(sub.target, body)  # durable enqueue
            else:
                raise ValueError(f"unknown channel {sub.channel}")
            return True
        except ValueError as exc:              # misconfig: not retryable
            await dead_letter(sub, body, str(exc))
            return False
        except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
            if attempt < MAX_RETRIES - 1:
                await asyncio.sleep(BASE_DELAY * 2 ** attempt + random.uniform(0, 0.4))
                continue
            logger.error("dead-letter %s: %s", sub.watcher_id, exc)
            await dead_letter(sub, body, str(exc))
            return False
    return False

Step 5 — Record an audit trail on every send

Persist one immutable row per outcome — delivered, deduplicated, suppressed, or dead-lettered — carrying the idempotency key and the lineage_id that ties the alert back to its source feed. This is what makes a subscriber’s alert history reconstructable, and it is the same provenance discipline described under data lineage and provenance tracking.

import time, json

async def audit(db, *, key, watcher_id, parcel_id, outcome, lineage_id, detail=""):
    await db.execute(
        "INSERT INTO alert_audit "
        "(idem_key, watcher_id, parcel_id, outcome, lineage_id, detail, ts) "
        "VALUES ($1,$2,$3,$4,$5,$6,$7)",
        key, watcher_id, parcel_id, outcome, lineage_id, detail, time.time(),
    )

Now assemble the runner. It threads matching, deduplication, delivery, and audit into a single pass, honouring the suppression gate first so a reprocess batch is absorbed silently.

async def run_notifier(events, index, deduper, db, session):
    stats = {"sent": 0, "dup": 0, "suppressed": 0, "dead": 0}
    sem = asyncio.Semaphore(16)

    async def dead_letter(sub, body, reason):
        await audit(db, key=body["idem_key"], watcher_id=sub.watcher_id,
                    parcel_id=body["parcel_id"], outcome="dead_letter",
                    lineage_id=body["lineage_id"], detail=reason)
        stats["dead"] += 1

    async def handle(sub, ev, sev):
        key = idem_key(sub.watcher_id, ev)
        if ev.get("is_reprocess"):
            await deduper.prime(key)
            stats["suppressed"] += 1
            return
        if not await deduper.is_new(key):
            await audit(db, key=key, watcher_id=sub.watcher_id,
                        parcel_id=ev["parcel_id"], outcome="duplicate",
                        lineage_id=ev["lineage_id"])
            stats["dup"] += 1
            return
        body = {"idem_key": key, "parcel_id": ev["parcel_id"],
                "lineage_id": ev["lineage_id"], "severity": sev,
                "from": ev["old_zoning"], "to": ev["new_zoning"],
                "sent_at": time.time()}
        async with sem:
            ok = await deliver(session, sub, body, dead_letter)
        if ok:
            await audit(db, key=key, watcher_id=sub.watcher_id,
                        parcel_id=ev["parcel_id"], outcome="sent",
                        lineage_id=ev["lineage_id"])
            stats["sent"] += 1

    await asyncio.gather(*(handle(sub, ev, sev)
                           for sub, ev, sev in match_events(index, events)),
                         return_exceptions=True)
    logger.info("notifier run: %s", stats)
    return stats

Verification and testing

Prove the pipeline fixed both failure modes rather than trading one for the other.

Replaying the same run five times, with and without the gate Line chart across five replays of an identical run. Without an idempotency gate the number of notifications delivered to subscribers grows linearly, from 14 after the first replay to 70 after the fifth, because every replay re-delivers the same fourteen changes. With the gate, delivery stays flat at 14 for all five replays: the first run delivers, and every subsequent replay matches the stored keys and delivers nothing. Replaying the same run five times, with and without the gate 0 20 40 60 80 1 2 3 4 5 Replay number, identical input each time Notifications delivered (cumulative) no idempotency gate with the gate (parcel + change hash)
This is the test worth writing: run the pipeline twice on identical input and assert that the second run delivers zero notifications. The flat line is the property you want, and it is cheap to verify — whereas the sloping line is discovered by subscribers, who then filter your alerts into a folder they never read.
  • Replay the storm. Feed the same batch twice. The first pass sends N alerts; the second must send zero and count N duplicates. If the second pass sends anything, the idempotency key is including a volatile field — check that sent_at is not in idem_key.
  • Assert exactly-once under a race. Dispatch the same event from two concurrent workers against the real dedupe store. Exactly one is_new returns True; the audit ledger must contain one sent row for that key, never two. This is what SET NX EX buys you.
  • Confirm suppression primes without sending. Run a batch with is_reprocess=True, assert zero sends, then run the same events without the flag and assert they are now deduplicated — proving the reprocess pass seeded the store rather than merely discarding events.
  • Check the miss cannot hide. Point a subscriber’s webhook at an endpoint that always fails, run one material event, and assert it appears in the dead-letter store with a retries_exhausted reason and a matching audit row. A missing dead-letter row means an event can vanish.
  • Spatial correctness. Use an L-shaped AOI and a parcel that sits in the bounding box notch but outside the polygon; the matcher must not select that subscriber. This confirms the exact intersects runs after the STRtree query.

Failure recovery

When something breaks mid-run, the pipeline must stay reproducible and lossless.

  • Drain and replay the dead-letter store. Replay re-enters run_notifier through the same path, so a replayed event still deduplicates against anything delivered while the endpoint was down — you cannot double-send by replaying. Alert on dead-letter depth so a persistently broken subscriber is noticed, not silently accumulating.
  • Recover from a mid-batch crash. Because each send is gated by an atomic claim and audited on success, restarting the batch re-attempts only events whose key was never successfully claimed-and-sent; already-delivered events short-circuit at the gate. Persist a batch checkpoint so a restart resumes rather than re-scans the whole day.
  • Handle a poisoned event. An event with invalid geometry that crashes matching should be quarantined and skipped, not allowed to abort the batch — the same dead-letter discipline used across error handling and retry logic in the ingestion layer applies here to the notifier.
  • Reconcile against source. If a subscriber reports a missed change, join the audit ledger on lineage_id back to the detection run and the source feed; every alert’s fate — sent, duplicate, suppressed, dead-lettered — is recorded, so “did we ever process this?” always has an answer.

Frequently asked questions

Why does re-running the batch send duplicates when I already use SET NX?

Almost always because a volatile field leaked into the idempotency key. If sent_at, a request id, or a fetch timestamp is part of the hashed input, every re-emission produces a new key and nothing deduplicates. Hash only the fields that define the alert’s meaning — watcher, parcel, jurisdiction, old and new zoning, and effective date — and add volatile fields to the delivered body only.

How do I keep a full county re-publish from paging everyone?

Have the ingestion layer flag reprocess batches, and route those events to prime instead of is_new. Priming records the idempotency key so a later genuine change still deduplicates, but it sends nothing during the bulk re-crawl. Pair it with a rate-based circuit breaker that trips when a jurisdiction’s inbound event count spikes far above baseline.

Should I claim the idempotency key before or after delivering?

Before. Claim atomically with SET NX EX, then deliver. Claiming after delivery opens a race where two workers both send. The trade-off is that a delivery which then fails must be dead-lettered, never silently abandoned, or the claimed key would block all future retries of that event. The runner here dead-letters on exhaustion for exactly that reason.

What TTL should the idempotency key use?

Long enough to cover the longest interval over which a source might re-publish the same layer — a week is a reasonable default for monthly-or-faster municipal refreshes. Set it too short and a re-publish outside the window re-alerts; too long and a genuinely re-decided parcel inside the window is swallowed. Including the effective date in the key protects against the latter, so you can keep the TTL generous.