Modeling Code Violation Lifecycles in Python

20 min read 1 sub-topic

This guide is part of the Violation Tracking & Compliance Enforcement track, which covers what happens after a permit is issued and the built environment has to be held to the code it was approved under. A code-violation case is not a status flag on a row — it is a regulated process with statutory clocks, due-process guarantees, and an evidentiary record that a hearing officer may one day read line by line. This page models that process as an explicit, data-driven state machine so every transition is legal, logged, and reproducible.

Problem Statement and Scope

A code-enforcement case moves through a fixed sequence of legally meaningful states: a complaint is reported, an officer performs an inspection, a notice of violation is issued, a cure period runs, a re-inspection confirms compliance or does not, and the case either resolves or escalates toward a hearing and ultimately a lien against the property. Each arrow between those states is constrained by ordinance — you cannot issue a lien on a case that never had a notice served, and you cannot close a case whose cure period has not lapsed. When this logic is scattered across if case.status == "..." branches in views, three failure modes appear:

  • Illegal transitions. A case jumps from reported straight to escalated because a well-meaning script skipped the notice step. The municipality has now denied the property owner the statutory cure window, and any fine assessed downstream is legally void.
  • Lost clocks. A cure period silently expires with no re-inspection scheduled, so a genuinely non-compliant property sits unenforced for months — or, worse, the clock is miscalculated and the owner is penalized a day early.
  • Unreconstructable history. A property owner appeals, and the office cannot produce a tamper-evident record of when each state changed, who drove the transition, and what the reason was.

The people who feel these failures are concrete: code-enforcement officers who need the next legal action surfaced automatically; Python automation builders who must make case routing deterministic across residential, commercial, and nuisance-abatement categories; and compliance officers who defend the office’s decisions in an administrative hearing or a state audit.

The reason a finite-state machine is the right model — rather than a status column and a pile of conditionals — is that code enforcement is a due-process pipeline. Each state grants the property owner specific rights (to be notified, to a cure window, to a hearing) and each transition is a discrete legal act that can be challenged. Modeling the process as an explicit machine makes three properties fall out for free: the set of legal moves is enumerable and reviewable by non-programmers, illegal moves are impossible by construction rather than by discipline, and the full history is a replayable event log rather than a mutable field whose past values are gone. Those are exactly the properties an appeal stresses.

The inputs to this subsystem are an existing case in some current state, a requested transition (an event such as serve notice or pass re-inspection), the identity of the actor requesting it, and the current clock. The outputs are a new persisted state, a recomputed set of statutory deadlines, and an immutable audit entry — never a bare status write. Downstream, a lapsed cure period is exactly what triggers tracking escalation timelines for unresolved violations, the child guide that turns these state definitions into scheduled, business-day-aware deadline math.

Code-violation case lifecycle as a guarded finite-state machine A code-violation case begins in the Reported state. An officer inspection moves it to Inspected, which either issues a Notice (to Notice Issued) or dismisses the case as unfounded (to Dismissed, terminal). Notice Issued starts a statutory Cure Period. When the cure clock elapses the case enters Re-inspection. A passing re-inspection transitions to Resolved (terminal); a failing re-inspection or a lapsed cure period transitions to Escalated. Escalated schedules a Hearing. A compliant hearing outcome returns to Resolved; an adverse outcome records a Lien against the property (terminal). Every transition is guarded by an ordinance rule and a statutory deadline, and every transition writes an immutable audit entry. inspect serve clock starts clock lapses pass fail schedule adverse comply unfounded Reported intake / complaint Inspected officer findings Notice Issued served · due process Cure Period statutory clock running Re-inspection verify cure Escalated enforcement action Hearing administrative Lien recorded · terminal Resolved compliant · terminal Dismissed

Prerequisites

This implementation targets Python 3.10+ (the examples use match/case, StrEnum, and dataclass(slots=True)). The state machine core is dependency-free and framework-agnostic; the optional declarative example uses the transitions library, and persistence assumes a SQLAlchemy-style session but ports cleanly to Django’s ORM.

pip install "transitions>=0.9" "SQLAlchemy>=2.0" "structlog>=24.1" "python-dateutil>=2.9"

Environment assumptions:

  • A permit/property datastore keyed on a stable parcel identifier, so a violation attaches to the same parcel the rest of the platform tracks.
  • An authenticated actor on every mutating call. Who may drive which transition is governed by the same model described in implementing role-based access for clerk portals — a field officer may serve a notice, only a supervisor may waive a cure requirement, and only the clerk of record may file a lien.
  • A write path to an append-only log or SIEM for transition audit events, kept for the full state-mandated records-retention period.
  • A jurisdiction configuration that supplies cure-period lengths and business-day/holiday rules; the deadline arithmetic itself lives in the child guide on escalation timelines.

Step 1 — Define the States and the Data-Driven Transition Table

Anchor the model to legally meaningful states, and encode the allowed moves as data rather than branching code. A transition table is what makes the machine auditable: an ordinance change becomes a reviewed data change, not a code rewrite, and the same table can be rendered for a hearing officer who has never seen Python.

from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum


class CaseState(str, Enum):
    REPORTED = "reported"
    INSPECTED = "inspected"
    NOTICE_ISSUED = "notice_issued"
    CURE_PERIOD = "cure_period"
    RE_INSPECTION = "re_inspection"
    ESCALATED = "escalated"
    HEARING = "hearing"
    RESOLVED = "resolved"      # terminal — compliant
    DISMISSED = "dismissed"    # terminal — unfounded
    LIEN = "lien"              # terminal — recorded against property


class Event(str, Enum):
    INSPECT = "inspect"
    DISMISS = "dismiss"
    SERVE_NOTICE = "serve_notice"
    START_CURE = "start_cure"
    LAPSE = "lapse"            # cure clock elapsed → time to re-inspect
    PASS_REINSPECT = "pass_reinspect"
    FAIL_REINSPECT = "fail_reinspect"
    SCHEDULE_HEARING = "schedule_hearing"
    HEARING_COMPLY = "hearing_comply"
    HEARING_ADVERSE = "hearing_adverse"


# (current_state, event) -> next_state. The single source of truth for legality.
# Every legal move in the ordinance appears here exactly once.
TRANSITIONS: dict[tuple[CaseState, Event], CaseState] = {
    (CaseState.REPORTED, Event.INSPECT): CaseState.INSPECTED,
    (CaseState.INSPECTED, Event.DISMISS): CaseState.DISMISSED,
    (CaseState.INSPECTED, Event.SERVE_NOTICE): CaseState.NOTICE_ISSUED,
    (CaseState.NOTICE_ISSUED, Event.START_CURE): CaseState.CURE_PERIOD,
    (CaseState.CURE_PERIOD, Event.LAPSE): CaseState.RE_INSPECTION,
    (CaseState.RE_INSPECTION, Event.PASS_REINSPECT): CaseState.RESOLVED,
    (CaseState.RE_INSPECTION, Event.FAIL_REINSPECT): CaseState.ESCALATED,
    (CaseState.ESCALATED, Event.SCHEDULE_HEARING): CaseState.HEARING,
    (CaseState.HEARING, Event.HEARING_COMPLY): CaseState.RESOLVED,
    (CaseState.HEARING, Event.HEARING_ADVERSE): CaseState.LIEN,
}

TERMINAL: frozenset[CaseState] = frozenset(
    {CaseState.RESOLVED, CaseState.DISMISSED, CaseState.LIEN}
)

Because the table is exhaustive, any pair not present is illegal by construction — there is no “default” branch that could quietly let a case skip its notice. That property is the whole point: due process is enforced by the shape of the data.

Note the two terminal-but-different closed states. DISMISSED records a complaint an inspection found unfounded — it must be distinguishable from RESOLVED, because a dismissed case was never a violation, while a resolved case was one that got cured. Keeping them separate matters for reporting: a jurisdiction’s compliance metrics count cured violations very differently from unfounded complaints, and conflating them understates or overstates enforcement activity in the state rollup. Splitting the terminal states here is what lets the downstream reporting layer aggregate honestly.

Step 2 — Attach Guard Conditions to Each Transition

Legality is necessary but not sufficient. A move can be structurally valid yet still forbidden right now: you may not lapse a cure period before its statutory clock has actually run, and you may not serve a notice on a case whose inspection recorded no violation. Guards are small predicates, keyed to the same (state, event) pair, that inspect the live case and the clock before the move is allowed.

from collections.abc import Callable
from datetime import datetime, timezone


@dataclass(slots=True)
class ViolationCase:
    case_id: str
    parcel_id: str
    state: CaseState = CaseState.REPORTED
    violation_found: bool = False
    cure_deadline: datetime | None = None    # set when the notice is served
    reinspection_passed: bool | None = None
    history: list[dict] = field(default_factory=list)


# A guard returns None if the move is allowed, or a human-readable reason if not.
Guard = Callable[[ViolationCase, datetime], str | None]


def _has_violation(case: ViolationCase, now: datetime) -> str | None:
    return None if case.violation_found else "inspection recorded no violation"


def _cure_clock_elapsed(case: ViolationCase, now: datetime) -> str | None:
    if case.cure_deadline is None:
        return "no cure deadline set"
    if now < case.cure_deadline:
        remaining = case.cure_deadline - now
        return f"cure period still running ({remaining.days}d left)"
    return None


GUARDS: dict[tuple[CaseState, Event], Guard] = {
    (CaseState.INSPECTED, Event.SERVE_NOTICE): _has_violation,
    (CaseState.CURE_PERIOD, Event.LAPSE): _cure_clock_elapsed,
}

The _cure_clock_elapsed guard is deliberately conservative: it refuses to advance a case even one moment early. Computing that cure_deadline correctly — skipping weekends and municipal holidays, honoring tolling — is involved enough to warrant its own treatment in tracking escalation timelines for unresolved violations; here we simply consume the deadline the timeline layer produces.

Step 3 — Build the Transition Engine with an Audit Log

Wrap the table and guards in a single apply function that is the only code path allowed to mutate case.state. It resolves legality, runs the guard, records the actor and reason in a hash-chained history entry, then commits the new state. Nothing else in the codebase writes the status field.

import hashlib
import json
import structlog

log = structlog.get_logger("violation.fsm")


class IllegalTransition(Exception):
    """The (state, event) pair is not in the ordinance transition table."""


class GuardFailed(Exception):
    """The move is structurally legal but blocked by a live condition."""


def apply(case: ViolationCase, event: Event, actor: str,
          reason: str = "", now: datetime | None = None) -> ViolationCase:
    now = now or datetime.now(timezone.utc)
    key = (case.state, event)

    dest = TRANSITIONS.get(key)
    if dest is None:
        raise IllegalTransition(f"{case.state.value} cannot {event.value}")

    guard = GUARDS.get(key)
    if guard is not None and (block := guard(case, now)) is not None:
        raise GuardFailed(f"{event.value} blocked: {block}")

    prev_hash = case.history[-1]["hash"] if case.history else "GENESIS"
    entry = {
        "ts": now.isoformat(), "actor": actor, "reason": reason,
        "from": case.state.value, "event": event.value, "to": dest.value,
        "prev": prev_hash,
    }
    entry["hash"] = hashlib.sha256(
        json.dumps(entry, sort_keys=True).encode()
    ).hexdigest()

    case.history.append(entry)
    case.state = dest
    log.info("transition", case_id=case.case_id, **entry)
    return case

Every state change now carries a timestamp, the actor, a reason, and a hash that chains to the previous entry — so a deleted or edited history row breaks the chain and is detectable during an appeal. This is the record a hearing officer or a state auditor actually reads.

Step 4 — Optionally Declare the Machine with transitions

The hand-rolled engine above is deliberately transparent, which is usually what you want for a regulated process. If you prefer a declarative model with callbacks, the transitions library expresses the same table and lets you hang conditions (guards) and after (side-effect) hooks directly on each edge. The behavior is identical; choose based on whether your team would rather read a table or a machine definition.

from transitions import Machine

DECLARATIVE = [
    {"trigger": "inspect", "source": "reported", "dest": "inspected"},
    {"trigger": "serve_notice", "source": "inspected", "dest": "notice_issued",
     "conditions": "has_violation"},
    {"trigger": "start_cure", "source": "notice_issued", "dest": "cure_period",
     "after": "arm_cure_clock"},
    {"trigger": "lapse", "source": "cure_period", "dest": "re_inspection",
     "conditions": "cure_clock_elapsed"},
    {"trigger": "pass_reinspect", "source": "re_inspection", "dest": "resolved"},
    {"trigger": "fail_reinspect", "source": "re_inspection", "dest": "escalated"},
    {"trigger": "schedule_hearing", "source": "escalated", "dest": "hearing"},
    {"trigger": "hearing_comply", "source": "hearing", "dest": "resolved"},
    {"trigger": "hearing_adverse", "source": "hearing", "dest": "lien"},
]


class CaseModel:
    def __init__(self, case: ViolationCase) -> None:
        self.case = case
        self.machine = Machine(
            model=self, states=[s.value for s in CaseState],
            transitions=DECLARATIVE, initial=case.state.value,
            auto_transitions=False,   # no implicit to_<state> shortcuts
            send_event=True,          # callbacks receive the EventData
        )

    def has_violation(self, event) -> bool:
        return self.case.violation_found

    def cure_clock_elapsed(self, event) -> bool:
        d = self.case.cure_deadline
        return d is not None and datetime.now(timezone.utc) >= d

    def arm_cure_clock(self, event) -> None:
        # Delegates to the escalation-timeline layer to compute the deadline.
        log.info("cure_armed", case_id=self.case.case_id)

Set auto_transitions=False so the library does not silently add a to_lien() shortcut that would bypass the whole ordinance path — the same fail-closed discipline the hand-rolled table gives you for free.

Step 5 — Persist State and Reconstruct History

The in-memory case must survive a process restart and be reconstructable field-by-field for an appeal. Persist the current state alongside the append-only history rows, and never UPDATE a history row — only insert. When you load a case, replay or verify the chain so a corrupted record surfaces at read time, not audit time.

from sqlalchemy import String, DateTime, JSON, Integer
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session


class Base(DeclarativeBase):
    pass


class CaseRow(Base):
    __tablename__ = "violation_cases"
    case_id: Mapped[str] = mapped_column(String, primary_key=True)
    parcel_id: Mapped[str] = mapped_column(String, index=True)
    state: Mapped[str] = mapped_column(String, index=True)
    cure_deadline: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))


class TransitionRow(Base):
    __tablename__ = "violation_transitions"   # append-only; never UPDATE/DELETE
    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    case_id: Mapped[str] = mapped_column(String, index=True)
    entry: Mapped[dict] = mapped_column(JSON)   # the hash-chained audit entry


def commit_transition(session: Session, case: ViolationCase) -> None:
    row = session.get(CaseRow, case.case_id)
    row.state = case.state.value
    row.cure_deadline = case.cure_deadline
    session.add(TransitionRow(case_id=case.case_id, entry=case.history[-1]))
    session.commit()   # state change and its audit row commit atomically


def verify_chain(rows: list[dict]) -> bool:
    prev = "GENESIS"
    for r in rows:
        expected = {k: v for k, v in r.items() if k != "hash"}
        if r["prev"] != prev:
            return False
        if hashlib.sha256(json.dumps(expected, sort_keys=True).encode()).hexdigest() != r["hash"]:
            return False
        prev = r["hash"]
    return True

Committing the state change and its audit row in one transaction is what prevents a state that has no corresponding history entry — the split-brain condition that makes an appeal indefensible.

Configuration Reference

Parameter Type Default Municipal-context notes
initial_state CaseState REPORTED Every case starts at intake; never construct a case in a mid-lifecycle state.
auto_transitions bool False With transitions, disable implicit to_<state>() moves that bypass the ordinance path.
guard_on_lapse Guard _cure_clock_elapsed Refuses to advance a case before its statutory cure clock has fully run.
require_reason_states set[CaseState] {ESCALATED, LIEN, DISMISSED} Transitions into consequential states must carry a non-empty justification.
terminal_states frozenset {RESOLVED, DISMISSED, LIEN} No outbound transitions; a reopen must create a linked new case, not mutate a closed one.
audit_sink str "append_only" append_only or siem; never a mutable table.
chain_seed str "GENESIS" First prev_hash; persist and reload the chain head so restarts do not fragment it.
hearing_required_before_lien bool True A lien can only follow a HEARING; enforce structurally, not by convention.

Error Handling and Edge Cases

Code enforcement breaks in specific, recurring ways. Handle each explicitly rather than letting it surface as a 500 or, worse, a silent illegal state.

  • Illegal transition requested. A batch job tries to escalate a case still in NOTICE_ISSUED. apply raises IllegalTransition; catch it, log a high-severity event, and surface the offending (state, event) pair. Never fall through to a default that mutates state.
  • Guard blocks a legal move. An officer tries to lapse a cure period a day early because their clock was wrong. GuardFailed returns the remaining time; show it, do not raise a 500. The deadline math that feeds this guard is the subject of the escalation-timeline guide.
  • Duplicate complaints on one parcel. Two residents report the same nuisance. Deduplicate by (parcel_id, violation_code, open) before creating a case, and link the second complaint as evidence on the existing case rather than opening a parallel lifecycle that could reach conflicting terminal states.
  • Reopening a closed case. A property that was RESOLVED re-offends. Do not transition out of a terminal state — create a new case that references the prior one, preserving the closed record intact for the audit trail.
  • Failed re-inspection arriving from the field. A mobile re-inspection comes back non-compliant. That result is captured upstream when capturing field inspection results on mobile devices syncs to the case, and it drives the FAIL_REINSPECT event into ESCALATED — one of the most common triggers of the whole enforcement path.
  • Notice never served. A NOTICE_ISSUED transition happened in the system, but certified mail was returned undelivered, so due process is incomplete. Gate START_CURE on a confirmed-service flag set by automating violation notice generation and delivery; an unserved notice must not start the cure clock.

Testing and Verification

A regulated state machine is exactly the code that must be tested as data. Enumerate legal and illegal (state, event) pairs and assert the outcome, so an ordinance change can never silently open an illegal path.

import pytest
from datetime import timedelta


def _fresh() -> ViolationCase:
    return ViolationCase(case_id="V-1", parcel_id="0042", violation_found=True)


@pytest.mark.parametrize("event,expected", [
    (Event.INSPECT, CaseState.INSPECTED),
])
def test_legal_first_move(event, expected):
    case = _fresh()
    apply(case, event, actor="[email protected]")
    assert case.state is expected


def test_illegal_skip_raises():
    case = _fresh()
    with pytest.raises(IllegalTransition):
        apply(case, Event.SCHEDULE_HEARING, actor="[email protected]")


def test_cannot_lapse_cure_early():
    case = _fresh()
    case.state = CaseState.CURE_PERIOD
    case.cure_deadline = datetime.now(timezone.utc) + timedelta(days=3)
    with pytest.raises(GuardFailed):
        apply(case, Event.LAPSE, actor="[email protected]")


def test_audit_chain_is_intact():
    case = _fresh()
    apply(case, Event.INSPECT, actor="[email protected]")
    apply(case, Event.SERVE_NOTICE, actor="[email protected]", reason="IPMC 304")
    assert verify_chain(case.history)
    case.history[0]["reason"] = "tampered"   # mutate a past row
    assert not verify_chain(case.history)     # chain now fails

A passing legality matrix plus a chain-verification test gives you a regression net: change a transition or a guard and the suite tells you exactly which legal boundary moved.

Integration Notes

The violation lifecycle sits at the downstream end of the permitting platform and pulls from several upstream subsystems. Who may drive each transition is resolved by the shared role model in implementing role-based access for clerk portals, so the authority to serve a notice or file a lien tracks the same scopes used everywhere else. The FAIL_REINSPECT and re-inspection events are fed by field data captured through capturing field inspection results on mobile devices, tying the enforcement machine to the inspection track. Entering NOTICE_ISSUED hands off to automating violation notice generation and delivery, which renders and serves the legal document and reports back the confirmed-service flag the cure clock depends on. And the deadline that guards the LAPSE transition — along with the scheduled jobs that fire it automatically — is the whole subject of tracking escalation timelines for unresolved violations.

Frequently Asked Questions

Should I hand-roll the state machine or use a library like transitions?

Either works because both encode the same transition table; the choice is about who reads the code. A hand-rolled table plus an apply gate is maximally transparent and easy to render for a hearing officer, which is usually what a regulated process wants. The transitions library is worth it when you have many callbacks and prefer a declarative machine definition — just set auto_transitions=False so it cannot add implicit shortcuts that bypass the ordinance path.

How do I stop a case from skipping the statutory cure period?

Encode the moves as an exhaustive table so any pair not listed is illegal by construction, then guard the LAPSE transition on the cure clock actually having elapsed. There is no default branch to fall through, so a script cannot jump from NOTICE_ISSUED to ESCALATED; it must pass through CURE_PERIOD and wait out the statutory window, which is exactly what due process requires.

What belongs in the audit log versus the case record?

The case record holds the current state and the live fields (cure deadline, parcel, latest re-inspection result). The audit log holds an append-only, hash-chained entry for every transition — timestamp, actor, reason, from-state, event, and to-state. The current state is derivable from replaying the log, but you store it too for query speed; you never edit the log, and you commit the state change and its log entry in one transaction so they can never disagree.

How should I handle a resolved case that re-offends?

Do not transition out of a terminal state. Create a new case linked to the prior one by parcel and prior-case reference, so the closed record stays intact and immutable for the audit trail while the new violation runs its own fresh lifecycle. Reopening in place would rewrite a legally settled record and destroy the very history an appeal depends on.

Explore deeper

Sub-topic Tracking Escalation Timelines for Unresolved Violations