Model/infrastructure/postgres/property_override_table.py
2026-06-05 12:18:13 +00:00

78 lines
2.9 KiB
Python

"""SQLModel mirror of the FE-owned ``property_overrides`` Drizzle table.
The schema + migration (``0221``) live in the ``assessment-model`` TS repo
(`src/app/db/schema/property_overrides.ts`); this row class only mirrors the
columns so the ``bulk_upload_finaliser`` Lambda can write the per-Property fact
layer at Finalise. Shape decided in ADR-0005; population (the join + resolve)
in ADR-0006.
"""
from __future__ import annotations
from datetime import datetime, timezone
from typing import ClassVar
from uuid import UUID, uuid4
from sqlalchemy import BigInteger, Column, SmallInteger, UniqueConstraint
from sqlalchemy import Enum as SAEnum
from sqlmodel import Field, SQLModel
# Mirror of the FE-owned ``override_component`` pgEnum (property_overrides.ts).
# The values ARE the classifier category keys used on both sides, so the
# finaliser maps category → component with no translation. A single SAEnum
# instance so test ``create_all`` emits one ``CREATE TYPE``; prod owns the type
# via Drizzle.
override_component_sa_enum = SAEnum(
"wall_type",
"roof_type",
"property_type",
"built_form_type",
name="override_component",
)
class PropertyOverrideRow(SQLModel, table=True):
"""Mirror of the FE-owned ``property_overrides`` table — one row per
``(property, override_component, building_part)`` carrying a denormalised
snapshot of the resolved enum (``override_value``) plus the raw cell text it
resolved from (``original_spreadsheet_description``)."""
__tablename__: ClassVar[str] = "property_overrides" # pyright: ignore[reportIncompatibleVariableOverride]
__table_args__: ClassVar[tuple[UniqueConstraint, ...]] = ( # pyright: ignore[reportIncompatibleVariableOverride]
UniqueConstraint(
"property_id",
"override_component",
"building_part",
name="property_overrides_property_component_part_unique",
),
)
id: UUID = Field(default_factory=uuid4, primary_key=True)
# bigint FKs in the Drizzle schema (→ property.id / portfolio.id, ON DELETE
# CASCADE). The FKs are enforced by the Drizzle migration, not declared here.
property_id: int = Field(
sa_column=Column(BigInteger, nullable=False, index=True),
)
portfolio_id: int = Field(
sa_column=Column(BigInteger, nullable=False),
)
# 0 = main building, 1 = extension 1, … (ADR-0004 ordering).
building_part: int = Field(
sa_column=Column(SmallInteger, nullable=False),
)
override_component: str = Field(
sa_column=Column(override_component_sa_enum, nullable=False),
)
override_value: str = Field(nullable=False)
original_spreadsheet_description: str = Field(nullable=False)
created_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
nullable=False,
)
updated_at: datetime = Field(
default_factory=lambda: datetime.now(timezone.utc),
nullable=False,
)