Data Governance & Privacy Compliance in Municipal Utility Billing

Consumption data quietly reveals when a household wakes, travels, and sleeps — which makes governing it a civic obligation, not a checkbox on a release form. This guide sits inside the broader Municipal Utility Billing Architecture & Rate Taxonomy and covers one workflow end to end: turning raw meter reads and customer records into a governed data flow where every field carries a sensitivity classification, every read is authorized, every retained record has an expiry, and every transformation is provable to an auditor. Get this layer right and the rating engine, reconciliation, and arrears routing all inherit a defensible source of truth. Get it wrong and a single leaked column or an un-purged eligibility record becomes a compliance finding — or a real harm to a resident.

# Problem Framing

In a production billing environment the governance failure is rarely a dramatic breach. It is subtler and more expensive: a legacy ORM that pulls customer_ssn and service_address into application memory during a routine rate recalculation and logs them at DEBUG; a low-income subsidy document that stays queryable three years after the household moved out because nothing enforced a retention window; a public-records export that ships exempt account-holder details alongside the aggregate consumption that was actually requested. Each of these is a policy violation that no unit test caught, because the data model treated sensitivity as documentation rather than as an enforced contract.

The remedy is engineering discipline applied at four seams that a billing pipeline crosses on every cycle. At ingestion, each field must declare its sensitivity so that validation, masking, and retention can act on it automatically. At access, the caller’s role must determine which columns even reach memory — least privilege enforced in code, not in a runbook. At rest, every record must carry a retention clock so that expired personal data is purged deterministically rather than accumulating indefinitely. At audit, every access and transformation must land in a tamper-evident log so a finance team can reconstruct exactly who saw what, when, and why. This page implements all four in Python, and cross-links the deeper drill-downs — Securing Customer PII in Utility Databases for the database layer and Security Boundaries & Role-Based Access for the API surface.

# Prerequisites

This workflow targets a modern, reproducible Python stack. Pin the following before implementing anything below:

• Python 3.11+ — required for zoneinfo in the standard library, StrEnum, and X | None union syntax used throughout.
• Pydantic v2 (pydantic>=2.6) — every governed schema uses v2 field_validator / model_validator semantics and Annotated field metadata; v1 models will not carry the sensitivity tags this page relies on.
• decimal.Decimal for all monetary and threshold arithmetic — never float. Consumption charges, subsidy amounts, and tolerance thresholds are compared and quantized as Decimal so governance decisions are reproducible across runs.
• zoneinfo (not pytz) — retention windows, grace periods, and purge cutoffs are computed in the utility’s local civil time, so every date comparison must be timezone-aware.
• Access assumptions — the calling service authenticates the operator and passes a validated role. Governance code enforces authorization; it does not perform authentication.

Throughout, UTILITY_TZ = ZoneInfo("America/Chicago") stands in for the municipality’s civil timezone; substitute the correct zone for your jurisdiction.

# Architecture Overview

The governance pipeline is a gate in front of the rating engine, not a report generated after it. A record is classified and validated before it enters memory, projected and masked according to the caller’s role, checked against its retention clock, and — whatever the outcome — written to an append-only audit chain.

Figure: The governance gate — classify, authorize, retention-check, and audit every record before it reaches the rating engine.

# Step-by-Step Implementation

Each step maps to one concrete billing operation. The code blocks compose into a single governance module; later steps assume the models and enums defined earlier.

# Step 1 — Declare a sensitivity contract at ingestion

Governance begins by making sensitivity a property of the data model itself. Each field is annotated with a classification, so masking and retention logic can act on metadata rather than a hand-maintained list of column names that inevitably drifts. This is the same schema-validation discipline used across the Schema Validation & Data Quality Checks pipeline, extended with a governance tag.

from __future__ import annotations

import logging
from datetime import datetime
from decimal import Decimal
from enum import StrEnum
from typing import Annotated
from zoneinfo import ZoneInfo

from pydantic import BaseModel, Field, field_validator

UTILITY_TZ = ZoneInfo("America/Chicago")
logger = logging.getLogger("utility_billing.governance")


class Sensitivity(StrEnum):
    PUBLIC = "public"          # aggregate, non-identifying
    INTERNAL = "internal"      # account IDs, consumption
    PII = "pii"                # name, service address, contact
    RESTRICTED = "restricted"  # SSN, payment tokens, subsidy docs


def classify(level: Sensitivity):
    """Attach a sensitivity level to a Pydantic field via json_schema_extra."""
    return Field(json_schema_extra={"sensitivity": level.value})


class CustomerRecord(BaseModel):
    account_id: Annotated[str, classify(Sensitivity.INTERNAL)] = Field(
        pattern=r"^ACC-\d{8}$"
    )
    consumption_kwh: Annotated[Decimal, classify(Sensitivity.INTERNAL)] = Field(ge=0)
    account_holder_name: Annotated[str, classify(Sensitivity.PII)]
    service_address: Annotated[str, classify(Sensitivity.PII)]
    ssn: Annotated[str | None, classify(Sensitivity.RESTRICTED)] = None
    subsidy_doc_id: Annotated[str | None, classify(Sensitivity.RESTRICTED)] = None
    read_timestamp: Annotated[datetime, classify(Sensitivity.INTERNAL)]

    @field_validator("consumption_kwh")
    @classmethod
    def reject_anomalous(cls, v: Decimal) -> Decimal:
        # A read above the municipal validation ceiling is quarantined, not billed.
        if v > Decimal("50000"):
            raise ValueError("consumption exceeds validation threshold; quarantine")
        return v

    @field_validator("read_timestamp")
    @classmethod
    def require_tz_aware(cls, v: datetime) -> datetime:
        # Naive timestamps make retention math ambiguous across DST — reject them.
        if v.tzinfo is None:
            raise ValueError("read_timestamp must be timezone-aware")
        return v.astimezone(UTILITY_TZ)

# Step 2 — Validate at the boundary and route failures to fallback

No record enters the pipeline unvalidated. A failure never defaults to an arbitrary historical average; it is quarantined and handed to Fallback Routing for Missing Rate Data so a provisional, clearly-flagged charge can be produced without corrupting the ledger.

from pydantic import ValidationError


def ingest(raw: dict, audit: list[dict]) -> CustomerRecord | None:
    """Validate one raw record. Valid records advance; invalid ones quarantine."""
    try:
        record = CustomerRecord.model_validate(raw)
    except ValidationError as exc:
        # Log the account id only — never echo the raw payload, which may hold PII.
        acct = raw.get("account_id", "UNKNOWN")
        logger.warning("validation failed for %s: %s", acct, exc.error_count())
        audit.append({
            "account_id": acct,
            "action": "quarantine",
            "reason": "schema_validation",
            "at": datetime.now(UTILITY_TZ).isoformat(),
        })
        return None  # caller invokes fallback routing for this account
    return record

# Step 3 — Enforce role-based field projection and masking

The caller’s role decides which fields reach memory. A billing clerk sees balances; only a privacy officer sees restricted fields. Rather than trust every query to remember the rule, projection is driven by the sensitivity tags declared in Step 1 — the enforcement point mirrored at the API layer by Security Boundaries & Role-Based Access.

# Highest sensitivity each role may see in the clear.
ROLE_CLEARANCE: dict[str, set[Sensitivity]] = {
    "billing_clerk": {Sensitivity.PUBLIC, Sensitivity.INTERNAL},
    "finance_analyst": {Sensitivity.PUBLIC, Sensitivity.INTERNAL, Sensitivity.PII},
    "privacy_officer": set(Sensitivity),  # all levels
}


def _field_sensitivity(model: type[BaseModel]) -> dict[str, Sensitivity]:
    out: dict[str, Sensitivity] = {}
    for name, info in model.model_fields.items():
        extra = info.json_schema_extra or {}
        out[name] = Sensitivity(extra.get("sensitivity", Sensitivity.INTERNAL))
    return out


def project(record: CustomerRecord, role: str, audit: list[dict]) -> dict:
    """Return a role-appropriate view; fields above clearance are masked, not dropped."""
    clearance = ROLE_CLEARANCE.get(role, {Sensitivity.PUBLIC})
    tags = _field_sensitivity(type(record))
    view, masked = {}, []
    for name, value in record.model_dump().items():
        if tags[name] in clearance or value is None:
            view[name] = value
        else:
            view[name] = "***REDACTED***"  # preserve shape; hide content
            masked.append(name)
    audit.append({
        "account_id": record.account_id,
        "action": "access",
        "role": role,
        "masked_fields": masked,
        "at": datetime.now(UTILITY_TZ).isoformat(),
    })
    return view

# Step 4 — Apply timezone-aware retention and purge

Personal data has an expiry. Financial records are retained under statute; the identifying fields attached to them are not. Retention windows are computed in civil time so that a purge cutoff lands on the correct calendar day regardless of DST. Restricted eligibility documents governed by the Assistance Program Eligibility Taxonomy carry the shortest windows.

from dataclasses import dataclass
from datetime import timedelta

# Retention (in days) by sensitivity level, per municipal record-retention policy.
RETENTION_DAYS: dict[Sensitivity, int] = {
    Sensitivity.PUBLIC: 3650,
    Sensitivity.INTERNAL: 2555,   # ~7 years for financial audit
    Sensitivity.PII: 1095,        # 3 years after last activity
    Sensitivity.RESTRICTED: 365,  # 1 year after eligibility lapse
}


@dataclass(frozen=True)
class RetentionOutcome:
    account_id: str
    tombstoned_fields: list[str]
    retained: bool


def apply_retention(
    record: CustomerRecord,
    last_activity: datetime,
    now: datetime,
    audit: list[dict],
) -> RetentionOutcome:
    """Tombstone fields whose retention window has closed; keep the ledger skeleton."""
    tags = _field_sensitivity(type(record))
    tombstoned: list[str] = []
    for name, level in tags.items():
        cutoff = last_activity.astimezone(UTILITY_TZ) + timedelta(
            days=RETENTION_DAYS[level]
        )
        if now.astimezone(UTILITY_TZ) >= cutoff and level in (
            Sensitivity.PII,
            Sensitivity.RESTRICTED,
        ):
            tombstoned.append(name)
    audit.append({
        "account_id": record.account_id,
        "action": "retention_purge",
        "tombstoned_fields": tombstoned,
        "at": now.astimezone(UTILITY_TZ).isoformat(),
    })
    # Internal/financial fields survive for audit; identifying fields are cleared.
    return RetentionOutcome(record.account_id, tombstoned, retained=bool(tombstoned) is False)

# Step 5 — Seal every decision into a hash-chained audit trail

Each preceding step appended to audit. Governance is only defensible if those records cannot be altered undetected, so the trail is chained: each record folds the previous record’s hash into its own SHA-256 digest over canonical JSON. Editing any past record breaks every hash after it. This is the same audit-logging pattern reconciliation relies on downstream.

import hashlib
import json


def seal_chain(events: list[dict], genesis: str = "GENESIS") -> list[dict]:
    """Return the events with a chained record_hash on each, in order."""
    prev = genesis
    sealed: list[dict] = []
    for event in events:
        payload = {**event, "prev_hash": prev}
        encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
        payload["record_hash"] = hashlib.sha256(encoded).hexdigest()
        prev = payload["record_hash"]
        sealed.append(payload)
    return sealed

# Edge-Case Handling

Municipal governance produces conflicts that generic data pipelines never anticipate. Handle each one deliberately:

• Erasure request versus financial retention. A resident’s right-to-erasure request collides with the statutory duty to retain billing history. Resolve it by tombstoning identifying fields (Step 4) while keeping the de-identified financial skeleton — never delete the ledger row, and record the erasure as its own audit event.
• Public-records request scoping. A request for aggregate consumption must not ship the PII attached to the same rows. Run the export through project with a PUBLIC/INTERNAL clearance so exempt fields are masked before the file ever leaves the process.
• Backfilled reads on purged accounts. A late AMI read arrives for an account whose PII was already tombstoned. Do not resurrect the identifying fields; attach the read to the retained financial skeleton and flag it for review rather than re-hydrating personal data.
• DST and leap-year retention boundaries. Because retention math uses zoneinfo and civil dates, a cutoff that falls on the spring-forward hour or on February 29 still resolves to the correct calendar day — naive UTC offsets would drift a purge by an hour and occasionally a day.
• Restricted fields in application logs. The most common leak is not a query but a stray logger.debug(record). Step 2 logs only the account id and an error count; enforce this as a rule, and scan logs for the redaction sentinel in verification.
• Role escalation mid-cycle. If an operator’s role changes during a long batch, projection must read the role per-record, not once at batch start, so a revoked clearance takes effect immediately.

# Verification and Audit Trail

A governance decision is only trustworthy if it can be replayed. Verify the chain integrity, confirm masking actually redacts, and assert that no restricted value escaped into a projected view.

def verify_chain(sealed: list[dict], genesis: str = "GENESIS") -> bool:
    """Recompute the hash chain; any in-place edit makes this return False."""
    prev = genesis
    for rec in sealed:
        stored = rec["record_hash"]
        payload = {k: v for k, v in rec.items() if k != "record_hash"}
        payload["prev_hash"] = prev
        encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
        if hashlib.sha256(encoded).hexdigest() != stored:
            return False
        prev = stored
    return True


def test_clerk_never_sees_restricted() -> None:
    audit: list[dict] = []
    record = CustomerRecord(
        account_id="ACC-00010001",
        consumption_kwh=Decimal("1200"),
        account_holder_name="Jane Roe",
        service_address="14 Elm St",
        ssn="000-00-0000",
        read_timestamp=datetime(2026, 3, 8, 2, 30, tzinfo=UTILITY_TZ),
    )
    view = project(record, "billing_clerk", audit)
    assert view["ssn"] == "***REDACTED***"
    assert view["account_holder_name"] == "***REDACTED***"
    assert view["consumption_kwh"] == Decimal("1200")  # in-clearance, in the clear
    assert audit[-1]["masked_fields"] == ["account_holder_name", "service_address", "ssn"]

Before a production run, replay the governance module against a fixture of historical accounts and assert the audit chain seals and re-verifies, and that the masked-field counts match the prior cycle within tolerance. At month-end the governed records feed batch reconciliation and ledger synchronization: automated scripts compare provisional (fallback-flagged) charges against verified reads and hold discrepancies beyond the statutory tolerance in suspense, so no un-audited adjustment reaches the general ledger. Every one of those comparisons cites the sealed audit chain, which is what satisfies a municipal or PUC audit request.

# Troubleshooting

# FAQ

# How do I reconcile a right-to-erasure request with billing-history retention?

Tombstone the identifying fields — name, address, SSN, subsidy document id — while keeping the de-identified financial row that statute requires you to retain. The erasure itself is recorded as an audit event, so you can later prove both that the personal data was removed and that the ledger was preserved. Never delete the row outright; that destroys the financial audit trail you are obligated to keep.

# Where should field-level sensitivity live — the database or the code?

Both, and they must agree. The authoritative classification lives on the schema (the sensitivity tag in Step 1) so masking, projection, and retention all read from one source. Enforce it a second time at the database with row- and column-level controls, covered in Securing Customer PII in Utility Databases. Application-layer masking prevents accidental exposure; database-layer controls stop a compromised or mis-scoped query.

# What is the difference between masking and purging?

Masking hides a value at access time for a caller who lacks clearance — the data still exists and a privacy officer can see it. Purging removes the value permanently once its retention window closes. Masking is an authorization decision made per read; purging is a lifecycle decision made per record. A field can be masked for years and then purged; the two are independent controls.

# Why compute retention windows in local time instead of UTC?

Retention policy is written in calendar days (“three years after last activity”), and calendar days are a civil-time concept. Computing a cutoff in UTC drifts it by the local offset and, twice a year across a DST boundary, by a full day — enough to purge a record a day early or hold it a day too long. Using zoneinfo against the utility’s civil timezone keeps every cutoff on the intended calendar date.

# How does this layer prove compliance to an auditor?

Every validation, access, mask, purge, and quarantine appends a record to a SHA-256 hash chain sealed in Step 5. An auditor recomputes the chain with verify_chain; if any historical record was altered, every subsequent hash changes and the check fails. Combined with the retention windows and role clearances, this gives a complete, tamper-evident answer to who saw what, when, why, and when it was removed.

# Related Topics

• Municipal Utility Billing Architecture & Rate Taxonomy — the parent guide; where governance sits in the wider billing stack.
• Securing Customer PII in Utility Databases — the database-layer drill-down: RLS, column projection, and secure connections.
• Security Boundaries & Role-Based Access — enforcing the same clearances at the API surface.
• Assistance Program Eligibility Taxonomy — the restricted eligibility data with the shortest retention windows.
• Fallback Routing for Missing Rate Data — where quarantined records go when validation fails.
• Schema Validation & Data Quality Checks — the ingestion-side validation this governance layer extends.

Up: Municipal Utility Billing Architecture & Rate Taxonomy