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:

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.

  • 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.