Lease Data Models

A lease data model is the contract that turns an unstructured legal document into a queryable, machine-readable asset that automated rent rolls, compliance alerts, and portfolio analytics can all trust. This page sits inside Core Architecture & Lease Taxonomy, and it is where every upstream stage converges: once clauses have been extracted and labeled, the data model decides how those facts are stored, versioned, and reconciled across a portfolio. Get the model wrong and a single mis-shaped escalation row can quietly post the wrong rent to thousands of leases; get it right and the rest of the platform becomes deterministic.

The specific workflow challenge this page solves: how do you persist a commercial lease — a moving target of amendments, conditional escalations, abatements, and jurisdictional quirks — in a schema that is strict enough to validate but flexible enough to survive the messiness of real abstraction output? The answer used in production lease platforms is a canonical record that decouples static lease metadata from time-bound financial and operational events, enforced at the ingestion boundary by a pydantic schema and an idempotent writer. Everything below is the engineering of that record.

Scope and where this fits

The data model is one stage in a longer chain, and keeping its responsibilities narrow is what makes it testable:

• Upstream of it: document ingestion, clause classification systems that assign typed labels, and the metadata normalization standards that guarantee dates, currencies, and entity names arrive in canonical form. By the time a payload reaches the model, every field should already be typed and normalized.
• This stage: validate the payload against a strict schema, resolve amendment precedence, assign a stable identity, and persist an auditable canonical record — or divert it to a dead-letter queue.
• Downstream of it: the stored record feeds the escalation formula mapping engine for rent computation, drives CAM charge reconciliation, and supplies the access-scoped views described in security & access boundaries.

A model that also tries to parse the escalation rate or classify a clause becomes impossible to reason about. It should answer one question — “is this a valid, identity-stable lease record, and what is its current effective state?” — and hand everything else off.

Prerequisites and environment setup

The reference implementation targets Python 3.11+ and a deliberately small dependency set so the model stays fast to import into any ingestion worker.

Install with pip install "pydantic>=2.6". Two environment assumptions matter. First, payloads arrive already normalized — string dates in ISO 8601, currencies as ISO 4217 codes, enums lowercased — because the model trusts the metadata normalization standards layer to have done that work; the model validates the contract, it does not repair it. Second, the store underneath can be relational or document-oriented, but it must support a temporal validity window per financial row so that amendment chains never overwrite history. The deeper schema-and-indexing treatment lives in how to structure a lease abstraction database for multi-property portfolios.

Canonical record shape

The model decomposes a lease into three concerns so each can be validated and versioned independently:

Enforcing a one-to-many relationship between a master lease record and its financial and clause children — rather than flattening everything into a wide row — is what lets amendments append instead of overwrite, and what keeps tenant_entity and premises_identifier from duplicating during rollups.

Primary implementation

The implementation below is the validation and ingestion seam: a strict pydantic v2 schema plus an idempotent writer that isolates failures instead of letting a single bad payload poison a batch.

import json
import logging
from datetime import datetime, date
from typing import Dict, Any, List, Optional
from decimal import Decimal
from pydantic import BaseModel, Field, field_validator, model_validator, ValidationError, ConfigDict
from enum import Enum

# Production logging configuration
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

class LeaseStatus(str, Enum):
    ACTIVE = "active"
    EXPIRED = "expired"
    TERMINATED = "terminated"
    HOLDING = "holding"

class EscalationTrigger(str, Enum):
    CPI = "cpi"
    FIXED = "fixed"
    MARKET = "market"
    NONE = "none"

class LeaseFinancials(BaseModel):
    model_config = ConfigDict(use_enum_values=True)
    base_rent: Decimal = Field(gt=0, decimal_places=2)
    currency: str = Field(pattern=r"^[A-Z]{3}$")          # ISO 4217
    payment_frequency: str = Field(pattern=r"^(monthly|quarterly|annually)$")
    security_deposit: Optional[Decimal] = Field(default=None, ge=0, decimal_places=2)
    escalation_type: EscalationTrigger = EscalationTrigger.NONE
    escalation_rate: Optional[float] = Field(default=None, ge=0, le=1.0)

    @model_validator(mode="after")
    def require_rate_for_rate_based_escalation(self) -> "LeaseFinancials":
        # A 'fixed' or 'cpi' escalation that carries no rate is silently wrong:
        # it would compound to zero and understate rent for the life of the lease.
        if self.escalation_type in {EscalationTrigger.FIXED, EscalationTrigger.CPI} \
                and self.escalation_rate is None:
            raise ValueError(f"escalation_type '{self.escalation_type}' requires escalation_rate")
        return self

class LeaseClause(BaseModel):
    clause_id: str = Field(pattern=r"^CLS-\d{4}$")
    clause_type: str
    effective_date: date
    raw_text: Optional[str] = None      # preserved verbatim for audit
    is_active: bool = True

class LeaseRecord(BaseModel):
    model_config = ConfigDict(strict=True)   # block implicit coercion that masks data-quality issues
    lease_id: str = Field(pattern=r"^LSE-\d{6}$")
    tenant_entity: str
    premises_identifier: str
    commencement_date: date
    expiration_date: date
    status: LeaseStatus
    financials: LeaseFinancials
    clauses: List[LeaseClause] = Field(default_factory=list)

    @field_validator("expiration_date")
    @classmethod
    def validate_dates(cls, v: date, info: Any) -> date:
        commencement = info.data.get("commencement_date")
        if commencement is not None and v <= commencement:
            raise ValueError("Expiration date must strictly follow commencement date")
        return v


def ingest_lease_data(raw_payload: Dict[str, Any]) -> Dict[str, Any]:
    """
    Production-grade ingestion seam for normalized lease abstraction output.
    Handles schema validation, ISO 8601 date coercion, and error isolation so a
    single malformed lease never fails an entire portfolio batch.
    """
    try:
        # Coerce ISO 8601 string dates to native date objects before strict validation.
        for field in ("commencement_date", "expiration_date"):
            if isinstance(raw_payload.get(field), str):
                raw_payload[field] = datetime.fromisoformat(raw_payload[field]).date()

        validated_record = LeaseRecord.model_validate(raw_payload)
        logger.info("Successfully ingested lease: %s", validated_record.lease_id)
        return {"status": "success", "data": validated_record.model_dump(mode="json")}

    except ValidationError as e:
        # Recoverable: bad data, not bad code. Route to the dead-letter queue with
        # the structured errors so the payload can be replayed after a fix.
        logger.error("Schema validation failed: %s", e.errors())
        return {"status": "error", "errors": e.errors()}
    except Exception as e:
        logger.critical("Unexpected ingestion failure: %s", str(e))
        return {"status": "critical_failure", "message": str(e)}


if __name__ == "__main__":
    sample_payload = {
        "lease_id": "LSE-100482",          # satisfies ^LSE-\d{6}$ (six digits)
        "tenant_entity": "Acme Logistics LLC",
        "premises_identifier": "BLDG-A-FL2-201",
        "commencement_date": "2024-01-01",
        "expiration_date": "2027-12-31",
        "status": "active",
        "financials": {
            "base_rent": 4500.00,
            "currency": "USD",
            "payment_frequency": "monthly",
            "security_deposit": 9000.00,
            "escalation_type": "fixed",
            "escalation_rate": 0.03
        },
        "clauses": [
            {"clause_id": "CLS-0012", "clause_type": "renewal_option",
             "effective_date": "2024-01-01", "is_active": True}
        ]
    }
    result = ingest_lease_data(sample_payload)
    print(json.dumps(result, indent=2))

The schema uses Decimal for every monetary field to prevent floating-point drift during rent roll calculations, and strict=True on the master record blocks the implicit coercion that would otherwise let a string "active" slip past a typo’d status enum. The model_validator on LeaseFinancials closes a real abstraction gap: a fixed or cpi escalation with a null rate is not an empty value, it is a silent error that would understate rent for the life of the lease.

Amendment precedence and temporal state

A lease is rarely the document you first abstracted. Renewals, rent commencement letters, and side amendments each supersede part of the base lease, and the data model — not the application layer — is where precedence must be resolved deterministically. The pattern is to store every financial obligation as a discrete row carrying its own validity window rather than mutating the original, so the current effective state is always the latest row whose window contains the as-of date.

from datetime import date
from typing import List, Optional
from pydantic import BaseModel
from decimal import Decimal

class RentSchedule(BaseModel):
    base_rent: Decimal
    effective_start: date
    effective_end: date = date(9999, 12, 31)   # open-ended until superseded
    source_document_id: str                     # which amendment produced this row

def effective_rent(schedules: List[RentSchedule], as_of: date) -> Optional[Decimal]:
    """Resolve the controlling rent for a point in time. Later effective_start wins
    on overlap, so an amendment cleanly supersedes the base lease without deleting it."""
    candidates = [s for s in schedules if s.effective_start <= as_of <= s.effective_end]
    if not candidates:
        return None
    return max(candidates, key=lambda s: s.effective_start).base_rent

Keeping superseded rows in place — effective_end defaulted to a sentinel far-future date — gives you point-in-time portfolio snapshots and a complete audit trail for free, which is exactly what compliance reviews and reconciliations demand.

Validation and quality gates

Schema validity is necessary but not sufficient. Three additional gates keep the model trustworthy at portfolio scale:

• Idempotent writes. Ingestion runs re-fire — on retry, on replay, on duplicate S3 events. Key every write on a deterministic hash of lease_id plus the source document version so a re-ingested payload upserts the same row instead of creating a phantom duplicate. Never rely on auto-increment identity for lease records.
• Confidence-aware admission. Records assembled from low-confidence clause classification should not enter the canonical store as if they were ground truth. Carry the upstream confidence forward and, below threshold, divert the record through fallback routing logic to a manual-review queue rather than admitting it silently.
• Dead-letter routing. A ValidationError is a data problem, not a code problem, so ingest_lease_data returns it as a structured payload rather than raising. Persist these to a dead-letter queue with the original input intact; once the schema or the upstream normalization is fixed, the batch replays cleanly.

Troubleshooting

Float drift in rent totals. Symptom: portfolio rent rollups disagree with the per-lease sum by fractions of a cent that compound across thousands of leases. Cause: a monetary field declared as float somewhere upstream. Fix: type every money field as Decimal end to end, and reject incoming floats at the boundary rather than casting them — casting 0.1 + 0.2 to Decimal preserves the error.

ValidationError on a valid-looking lease_id. Symptom: identifiers like LSE-12345 are rejected. Cause: the pattern ^LSE-\d{6}$ requires exactly six digits; a five-digit id fails. Fix: confirm the upstream id mint zero-pads to six, or relax the pattern deliberately — do not loosen it accidentally by switching to \d+, which would let malformed ids through.

Amendment overwrites the base lease. Symptom: historical rent disappears after an amendment is ingested. Cause: the writer mutated the existing financial row instead of appending a new one with its own validity window. Fix: make rent schedules append-only and resolve current state with the effective_rent lookup above; never UPDATE a superseded row’s value.

strict=True rejects a normalized payload. Symptom: a record fails validation even though the values look correct. Cause: strict mode refuses implicit coercion — an integer where a Decimal is expected, or a date object passed as a string. Fix: do the explicit coercion (as the date loop does) before model_validate, and ensure the normalization layer emits the exact types the schema declares.

Escalation silently computes to zero. Symptom: rent never escalates despite a fixed escalation type. Cause: escalation_rate arrived null and was admitted. Fix: the model_validator shown above rejects rate-based escalations with a null rate; ensure it runs and that the dead-letter queue surfaces these for re-abstraction.

Duplicate lease rows after a retry. Symptom: the same lease appears twice after an ingestion replay. Cause: non-idempotent writes keyed on auto-increment identity. Fix: upsert on a deterministic key derived from lease_id and source document version.

Performance and scale notes

For multi-property portfolios, the cost is rarely validation itself — pydantic v2’s Rust core validates tens of thousands of records per second — but the surrounding I/O. Batch writes in chunks of a few hundred records inside a single transaction so the database commits amortize, and stream large portfolios rather than materializing the whole payload set in memory. When ingestion is event-driven, run validation in worker processes and push only validated, serialized records onto the write path, keeping the heavy clause text out of hot in-memory structures by storing raw_text once and referencing it by id. The same async patterns that scale extraction apply here; coordinating them across stages is covered in the parsing-side workflows.

Frequently asked questions

Related

• How to Structure a Lease Abstraction Database for Multi-Property Portfolios — the temporal relational schema, JSONB clause storage, and indexing strategy beneath this model.
• Clause Classification Systems — the upstream stage that produces the typed clauses this model stores.
• Escalation Formula Mapping — how a stored financial row is turned into an actual computed rent.
• Metadata Normalization Standards — the normalization contract that guarantees fields arrive in the types this schema expects.
• Fallback Routing Logic — where low-confidence and schema-invalid records are diverted for review.

← Back to Core Architecture & Lease Taxonomy

← Back to Core Architecture & Lease Taxonomy