Building Automated Inspection Notification Systems

19 min read 1 sub-topic

This guide is part of the Inspection Scheduling & Field Operations track, which turns issued permits into scheduled site visits and field results. A notification system is the connective tissue: it tells contractors and applicants when an inspection is booked, moved, cancelled, or graded, across every channel they actually read.

Problem Statement and Scope

An inspection schedule that nobody reliably receives is worse than no schedule at all — it produces missed visits, re-inspection fees, and disputes about whether notice was ever given. The moment a permit reaches the inspection stage of its lifecycle, at least three parties need to stay synchronized: the field inspector whose route is being planned, the contractor who must have the site ready and accessible, and the applicant or owner of record who is legally entitled to notice of a municipal action. When those notifications are sent by a clerk copy-pasting from a calendar, the failures are predictable — a reschedule email that never goes out, an SMS sent at 4 a.m., a duplicate blast when a batch job retries, or a “your inspection passed” message delivered to a phone number that changed two permits ago.

The people who feel this are concrete. Municipal clerks field the angry phone call when a contractor shows up on the wrong day. Python automation builders own the reliability of a fan-out that touches an email provider, an SMS gateway, and a calendaring subsystem, each with its own failure envelope. Compliance officers must later prove that statutory notice was delivered — with a timestamp, a channel, and a delivery status — when a re-inspection fee is contested.

The scope of this guide is the notification subsystem specifically: given an inspection event (scheduled, rescheduled, cancelled, or a posted result), fan it out to the right recipients over email, SMS, and calendar invites, with templating, delivery retries, idempotency so a retry never double-sends, quiet-hours suppression, opt-out handling, and a delivery-status ledger. The inputs are inspection events emitted by the scheduling engine that syncs inspector calendars with permit milestones; the outputs are dispatched, de-duplicated messages plus an auditable record of every send attempt and its outcome. Producing the subscribable calendar feed that rides alongside these messages is covered in the companion guide on generating iCal feeds for contractor inspection appointments.

Multi-channel inspection notification fan-out with dedup, preferences, and a delivery ledger An inspection event enters carrying a deterministic idempotency key. A dedup gate drops keys already recorded as sent. A preference resolver applies opt-out and quiet-hours rules per recipient and channel. A template renderer produces channel-specific bodies. The message then fans out to three channel adapters — email, SMS, and calendar invite — each backed by a retry queue with exponential backoff and jitter. Every send attempt writes an append-only row to a delivery-status ledger capturing channel, recipient, provider message id, and terminal status, which is the record compliance officers query to prove statutory notice. event unique allowed Inspection event + idempotency key Dedup gate drop seen keys Preferences opt-out quiet-hours Render per-channel body Email adapter retry · backoff SMS adapter retry · backoff Calendar invite iCalendar Append-only delivery-status ledger channel · recipient · provider message id · terminal status

Prerequisites

This implementation targets Python 3.10+ (the examples use match/case, StrEnum, and structural pattern matching on event types). It is transport-agnostic: the channel adapters wrap whatever email and SMS providers your jurisdiction has procured, and the calendar payload is standards-based iCalendar so it works with any client.

pip install "pydantic>=2.6" "jinja2>=3.1" "icalendar>=5.0" "twilio>=9.0" \
            "tenacity>=8.2" "structlog>=24.1" "SQLAlchemy>=2.0"

Environment assumptions:

  • Inspection events arrive from the scheduling engine as validated records that already carry a permit id, an inspector, an appointment window, and a recipient roster. The shape of those records should conform to the same JSON schemas for building permits the rest of the platform versions, so a notification never invents a field the record does not define.
  • A relational store for the delivery-status ledger and the opt-out/preference tables. These are consulted on the hot path, so index them on (recipient, channel).
  • Credentials for an email provider (SMTP relay or API) and an SMS provider. The examples use the Twilio client library for SMS to keep the code concrete; swap the adapter body for your procured gateway without touching the pipeline.
  • The recipient roster is guarded data. Reads and writes of contact details flow through the same access controls described in implementing role-based access for clerk portals — a phone number is PII and must not be logged in the clear.

Step 1 — Model the Notification Event and a Deterministic Idempotency Key

Everything downstream depends on one idea: a notification is uniquely identified by what happened, not by when the job ran. If the scheduler re-emits the same reschedule twice, or a worker crashes mid-fan-out and the task is retried, the recipient must not receive two texts. Encode that by deriving an idempotency key deterministically from the event’s stable attributes.

import hashlib
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, Field


class EventKind(str, Enum):
    SCHEDULED = "scheduled"
    RESCHEDULED = "rescheduled"
    CANCELLED = "cancelled"
    RESULT_POSTED = "result_posted"


class Channel(str, Enum):
    EMAIL = "email"
    SMS = "sms"
    CALENDAR = "calendar"


class Recipient(BaseModel):
    party_id: str            # stable id of the contractor / applicant
    email: str | None = None
    phone_e164: str | None = None   # E.164, e.g. +15125550142


class InspectionEvent(BaseModel):
    permit_id: str
    inspection_id: str
    kind: EventKind
    # The scheduled window; for CANCELLED this is the slot being released.
    starts_at: datetime
    ends_at: datetime
    sequence: int = 0        # bumped each time this inspection is revised
    recipients: list[Recipient] = Field(default_factory=list)

    def idempotency_key(self, recipient: Recipient, channel: Channel) -> str:
        # Stable across retries: same event + revision + recipient + channel
        # always hashes to the same key, so a redelivery is detectable.
        basis = f"{self.inspection_id}:{self.kind.value}:{self.sequence}:" \
                f"{recipient.party_id}:{channel.value}"
        return hashlib.sha256(basis.encode()).hexdigest()

The sequence field is what lets a genuine change through while still blocking a duplicate: a reschedule increments it, so its key differs from the original booking, but a retried delivery of that same reschedule reuses the key and is dropped. This is the identical revision counter you will stamp into the calendar feed’s SEQUENCE property when generating iCal feeds for contractor inspection appointments, keeping the message stream and the calendar in lockstep.

Step 2 — Resolve Recipients, Opt-Outs, and Quiet Hours

Before rendering anything, decide who is actually allowed to be contacted on which channel, and whether now is a permitted time. Municipal messaging is subject to consent rules (a contractor who opted out of SMS must not be texted) and to common-sense quiet hours (no automated call or text overnight), with one critical carve-out: a hard cancellation is time-sensitive and often overrides quiet hours because a contractor mobilizing a crew at dawn needs to know tonight.

from datetime import time, timezone
from zoneinfo import ZoneInfo


QUIET_START, QUIET_END = time(21, 0), time(7, 0)   # 9pm–7am local


def in_quiet_hours(now_local: datetime) -> bool:
    t = now_local.timetz()
    # Window straddles midnight, so OR the two halves.
    return t >= QUIET_START.replace(tzinfo=t.tzinfo) or \
        t < QUIET_END.replace(tzinfo=t.tzinfo)


def channels_for(event: InspectionEvent, r: Recipient,
                 opted_out: set[Channel], local_tz: str) -> list[Channel]:
    wanted: list[Channel] = []
    if r.email:
        wanted.append(Channel.EMAIL)   # email is not quiet-hours restricted
    if r.phone_e164:
        wanted.append(Channel.SMS)
    wanted.append(Channel.CALENDAR)    # a .ics attachment rides with email

    now_local = datetime.now(timezone.utc).astimezone(ZoneInfo(local_tz))
    urgent = event.kind is EventKind.CANCELLED
    allowed: list[Channel] = []
    for ch in wanted:
        if ch in opted_out:
            continue                    # honor consent unconditionally
        if ch is Channel.SMS and in_quiet_hours(now_local) and not urgent:
            continue                    # defer non-urgent SMS out of quiet hours
        allowed.append(ch)
    return allowed

Two rules are non-negotiable here. Opt-out is honored unconditionally — consent is never overridden by urgency, only quiet hours are. And quiet-hours suppression defers rather than drops: the caller should re-enqueue a suppressed non-urgent SMS for the next allowed window rather than silently losing the notice, so a contractor still learns about a booking made at 11 p.m., just at 7 a.m.

Step 3 — Render Channel-Specific Templates

Each channel has a different envelope: email tolerates a subject line and rich body, SMS is a terse single segment, and the calendar payload is a structured object. Keep the content in versioned templates so a wording change is a reviewed pull request, not a code edit, and so a compliance officer can point to exactly which template produced a given notice.

from jinja2 import Environment, DictLoader, select_autoescape

TEMPLATES = {
    "scheduled.email.subject": "Inspection scheduled for permit {{ permit_id }}",
    "scheduled.email.body":
        "Your {{ kind }} inspection for permit {{ permit_id }} is booked for "
        "{{ starts_at.strftime('%A %B %-d, %Y at %-I:%M %p') }}.\n"
        "Please ensure the site is accessible. Reply STOP to opt out of texts.",
    "scheduled.sms":
        "City inspection {{ permit_id }}: {{ starts_at.strftime('%b %-d %-I:%M%p') }}. "
        "Site must be accessible. Reply STOP to opt out.",
    "cancelled.sms":
        "City inspection {{ permit_id }} on "
        "{{ starts_at.strftime('%b %-d') }} is CANCELLED. Do not mobilize.",
}

_env = Environment(loader=DictLoader(TEMPLATES),
                   autoescape=select_autoescape(["html"]),
                   trim_blocks=True, lstrip_blocks=True)


def render(template_id: str, event: InspectionEvent) -> str:
    tmpl = _env.get_template(template_id)  # raises if the template is missing
    return tmpl.render(permit_id=event.permit_id, kind=event.kind.value,
                       starts_at=event.starts_at, ends_at=event.ends_at).strip()

Templating by convention ({kind}.{channel}[.subject]) means adding a channel or an event kind is additive. A missing template raises loudly at render time rather than sending a half-populated message — fail-closed, so a contractor never receives a message with an unrendered permit_id placeholder.

Step 4 — Dispatch Through Channel Adapters with Retries

Each adapter owns exactly one provider’s quirks and presents a uniform send() -> provider_message_id contract. Wrap transient provider failures in bounded, jittered retries; do not retry a permanent rejection (an invalid number, a hard bounce), which only burns quota and delays the ledger’s terminal state. The calendar “channel” does not call a remote provider — it attaches a generated .ics to the email — but it still records a delivery row.

from tenacity import (retry, stop_after_attempt, wait_exponential_jitter,
                      retry_if_exception_type)
from twilio.rest import Client as TwilioClient
from twilio.base.exceptions import TwilioRestException


class TransientSendError(Exception):
    """Retryable: timeouts, 429s, provider 5xx."""


class PermanentSendError(Exception):
    """Not retryable: invalid recipient, hard bounce, opt-out at provider."""


_sms = TwilioClient()  # reads credentials from the environment


@retry(retry=retry_if_exception_type(TransientSendError),
       stop=stop_after_attempt(5),
       wait=wait_exponential_jitter(initial=1, max=30), reraise=True)
def send_sms(to_e164: str, body: str, idem_key: str) -> str:
    try:
        msg = _sms.messages.create(
            to=to_e164, from_="+15125550100", body=body,
            # Provider-side idempotency: a retried create with the same key
            # returns the original message instead of sending twice.
            idempotency_key=idem_key,
        )
        return msg.sid
    except TwilioRestException as exc:
        if exc.status == 429 or 500 <= exc.status < 600:
            raise TransientSendError(str(exc)) from exc
        raise PermanentSendError(f"{exc.code}: {exc.msg}") from exc

Two layers of idempotency stack here. The local dedup gate (Step 5) stops a duplicate before it reaches the provider; the provider-side idempotency_key is the belt-and-suspenders guard for the window between a successful send and the ledger write, so a crash in that gap cannot double-charge the contractor a text. For batch-heavy rescheduling — a whole day’s route slipping because of weather — route the fan-out through the same durable queue patterns used for managing Celery task queues for overnight batch imports so a provider outage backs off instead of hammering.

Step 5 — Gate on the Ledger and Record Every Attempt

The delivery-status ledger is both the dedup source of truth and the compliance artifact. Claim the idempotency key transactionally before dispatch; if the row already exists in a terminal-sent state, skip. After the adapter returns, update the row with the provider message id and status. This ordering guarantees at-most-once user-visible delivery even under concurrent workers.

from sqlalchemy import String, DateTime, UniqueConstraint, select
from sqlalchemy.orm import Mapped, mapped_column, Session
from sqlalchemy.exc import IntegrityError


class DeliveryRow(Base):  # Base = declarative_base()
    __tablename__ = "notification_delivery"
    __table_args__ = (UniqueConstraint("idem_key", name="uq_idem"),)
    idem_key: Mapped[str] = mapped_column(String(64), primary_key=True)
    inspection_id: Mapped[str] = mapped_column(String, index=True)
    channel: Mapped[str] = mapped_column(String(16))
    recipient_ref: Mapped[str] = mapped_column(String)   # party_id, never raw PII
    status: Mapped[str] = mapped_column(String(16))      # queued|sent|failed|skipped
    provider_msg_id: Mapped[str | None] = mapped_column(String, nullable=True)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))


def claim(session: Session, key: str, event: InspectionEvent,
          ch: Channel, r: Recipient) -> bool:
    """Return True if we won the claim and should send; False if already handled."""
    row = DeliveryRow(idem_key=key, inspection_id=event.inspection_id,
                      channel=ch.value, recipient_ref=r.party_id,
                      status="queued", updated_at=datetime.now(timezone.utc))
    session.add(row)
    try:
        session.commit()          # unique constraint enforces the claim
        return True
    except IntegrityError:
        session.rollback()
        existing = session.get(DeliveryRow, key)
        return existing is not None and existing.status == "failed"  # retry only failures

Because the claim is a unique-constraint insert, two workers racing on the same event resolve deterministically: one wins and sends, the other sees the constraint violation and backs off. A previously failed row is the one case a retry is allowed to re-drive, which is how a transient outage eventually clears without a human. This append-then-update ledger is the exact record a compliance officer pulls to prove notice — and it mirrors how delivery is tracked when automating violation notice generation and delivery in the enforcement track, so both subsystems answer a “was legal notice served?” question the same way.

Configuration Reference

Parameter Type Default Municipal-context notes
quiet_hours_local tuple[time, time] (21:00, 07:00) Local wall-clock window; applies to SMS/voice, never to email or hard cancellations.
quiet_hours_override_kinds set[EventKind] {CANCELLED} Event kinds urgent enough to send during quiet hours.
max_send_attempts int 5 Per-channel retry ceiling for transient failures before terminal failed.
retry_backoff_initial_s float 1.0 Exponential base with jitter; caps provider hammering during an outage.
retry_backoff_max_s float 30.0 Upper bound per attempt; keeps a stuck queue from starving fresh events.
dedup_window n/a permanent Idempotency keys are retained for the record’s full retention period, not a TTL.
sms_from_number str — Jurisdiction’s registered E.164 sender; must match the number carriers approved.
pii_in_logs bool False Never log raw email/phone; log the party_id reference and hash the key.
opt_out_keyword str "STOP" Inbound STOP writes an opt-out row that Step 2 honors unconditionally.

Error Handling and Edge Cases

Municipal notification breaks in specific, recurring ways. Handle each explicitly rather than letting it surface as a silent non-delivery.

  • Duplicate fan-out on worker retry. A Celery task crashes after sending but before the ledger update; the retry re-runs the whole event. The deterministic idempotency key plus the provider-side key means the re-run is dropped at the claim and, worst case, deduped at the provider — the contractor sees one message.
  • Quiet-hours notice lost. A booking made at 11 p.m. suppresses the SMS and, if you only drop it, the contractor never learns. Re-enqueue suppressed non-urgent messages for the next allowed window instead of discarding them, and log the deferral.
  • Stale contact details. A phone number belongs to a prior owner. A hard bounce or an invalid-number rejection is a PermanentSendError — do not retry it; mark the row failed, and surface a stale-contact report to clerks so the roster is corrected at the source record.
  • Provider outage during a deadline surge. The SMS gateway 429s across a whole rescheduled route. Exponential backoff with jitter spreads the retries; pair the fan-out with a circuit breaker as in configuring circuit breakers for permit database timeouts so a dead provider trips fast instead of exhausting every worker.
  • Timezone and DST drift. Quiet-hours math done in UTC will text someone at 5 a.m. across a DST boundary. Always evaluate the window in the recipient’s local zone — the same class of bug covered in depth in handling timezone and DST edge cases in inspection scheduling.
  • Opt-out race. A contractor replies STOP while a batch is mid-flight. Check the opt-out table inside the claim transaction, not only at enqueue time, so a consent change lands before the send.

Testing and Verification

Notification logic must be tested as data — enumerate event kind × channel × time-of-day and assert the decision, so a wording or consent change can never silently widen who gets contacted when.

import pytest
from datetime import datetime, timezone


def test_duplicate_event_is_deduped(session):
    ev = InspectionEvent(permit_id="P-1", inspection_id="I-9",
                         kind=EventKind.SCHEDULED,
                         starts_at=datetime(2026, 8, 3, 14, tzinfo=timezone.utc),
                         ends_at=datetime(2026, 8, 3, 15, tzinfo=timezone.utc))
    r = Recipient(party_id="C-42", phone_e164="+15125550142")
    key = ev.idempotency_key(r, Channel.SMS)
    assert claim(session, key, ev, Channel.SMS, r) is True   # first wins
    assert claim(session, key, ev, Channel.SMS, r) is False  # replay dropped


def test_reschedule_bumps_key(session):
    base = dict(permit_id="P-1", inspection_id="I-9",
                starts_at=datetime(2026, 8, 3, 14, tzinfo=timezone.utc),
                ends_at=datetime(2026, 8, 3, 15, tzinfo=timezone.utc))
    r = Recipient(party_id="C-42", phone_e164="+15125550142")
    k0 = InspectionEvent(kind=EventKind.SCHEDULED, sequence=0, **base) \
        .idempotency_key(r, Channel.SMS)
    k1 = InspectionEvent(kind=EventKind.RESCHEDULED, sequence=1, **base) \
        .idempotency_key(r, Channel.SMS)
    assert k0 != k1   # a real change is not swallowed by dedup


@pytest.mark.parametrize("hour,urgent,expect_sms", [
    (23, False, False),   # quiet hours, non-urgent -> deferred
    (23, True, True),     # quiet hours, cancellation -> sent
    (10, False, True),    # daytime -> sent
])
def test_quiet_hours_gate(hour, urgent, expect_sms):
    now = datetime(2026, 8, 3, hour, tzinfo=ZoneInfo("America/Chicago"))
    kind = EventKind.CANCELLED if urgent else EventKind.SCHEDULED
    ev = InspectionEvent(permit_id="P-1", inspection_id="I-9", kind=kind,
                         starts_at=now, ends_at=now)
    r = Recipient(party_id="C-42", phone_e164="+15125550142")
    # Freeze "now" via your clock injection; illustrative here.
    got = Channel.SMS in channels_for(ev, r, set(), "America/Chicago")
    assert got is expect_sms or urgent  # cancellations always allowed

A passing dedup test plus a quiet-hours matrix gives you a regression net: change the suppression rule or the key basis and the suite tells you exactly which recipient boundary moved.

Integration Notes

This subsystem sits between scheduling and the field. It consumes appointment windows from the engine that syncs inspector calendars with permit milestones and emits the subscribable feed detailed in generating iCal feeds for contractor inspection appointments, so a contractor’s phone alert and their calendar entry describe the same revision of the same visit. When a result is posted from the field, the RESULT_POSTED event both notifies the applicant and, on a failed inspection, can seed the enforcement pipeline that handles automating violation notice generation and delivery. Upstream, the contact roster and permit context are produced and guarded by the Core Architecture & Code Taxonomy for Municipal Permits track, so a phone number is never read or logged outside the access model that governs it.

Frequently Asked Questions

How do I stop a job retry from sending the same contractor two texts?

Derive the idempotency key deterministically from stable event attributes — inspection id, event kind, revision sequence, recipient, and channel — and claim it with a unique-constraint insert before dispatch. A retried task recomputes the same key, loses the claim, and is dropped. Add the provider’s own idempotency key as a second layer for the narrow window between a successful send and the ledger write.

Should quiet hours ever be overridden?

Only for genuinely urgent, time-sensitive events — a hard cancellation is the canonical case, because a contractor mobilizing a crew at dawn needs to know tonight. Opt-out is never overridden by urgency; only quiet hours are. And a suppressed non-urgent message should be deferred to the next allowed window, not dropped, so notice still reaches the recipient.

How do I handle a contractor who replies STOP mid-batch?

Write an opt-out row on the inbound STOP and check that table inside the claim transaction, not only when the batch was enqueued. Because the opt-out is honored unconditionally at resolve time, a consent change that lands while a batch is in flight takes effect on the very next send rather than after the batch drains.

What do I keep to prove statutory notice was delivered?

The append-only delivery-status ledger: one row per channel per recipient per revision, carrying the channel, a party_id reference (never raw PII), the provider message id, the terminal status, and a timestamp. Retain it for the full state-mandated period. That row is what a compliance officer produces when a re-inspection fee or a missed-notice claim is contested.

Explore deeper

Sub-topic Generating iCal Feeds for Contractor Inspection Appointments