67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
"""Lookup ЗУ под изъятие/резервирование из ``land_reservation`` (#1091, #1062, #1067).
|
||
|
||
Читает land_reservation (м.136, наполняется reservation_ingest task) и возвращает
|
||
актуальные постановления по кад-номеру. Геометрия не хранится — join к cad_parcels.geom
|
||
делается на уровне analyze (follow-up после merge #1115).
|
||
|
||
Graceful: нет cad_num / таблица ещё не задеплоена (OperationalError/ProgrammingError) → [].
|
||
Зеркалит стиль ird_overlay_lookup.py.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.exc import OperationalError, ProgrammingError
|
||
from sqlalchemy.orm import Session
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_LOOKUP_SQL = text(
|
||
"""
|
||
SELECT reservation_kind, basis_act, act_number, act_date, purpose, doc_url
|
||
FROM land_reservation
|
||
WHERE cad_num = CAST(:cad_num AS text)
|
||
AND is_active
|
||
ORDER BY act_date DESC NULLS LAST, id DESC
|
||
"""
|
||
)
|
||
|
||
|
||
def parcel_reservations(db: Session, cad_num: str | None) -> list[dict[str, object]]:
|
||
"""Возвращает актуальные постановления об изъятии/резервировании по кад-номеру.
|
||
|
||
Args:
|
||
db: сессия SQLAlchemy.
|
||
cad_num: кадастровый номер ЗУ (например, «66:41:0101001:123»).
|
||
None → пустой список.
|
||
|
||
Returns:
|
||
Список словарей: [{reservation_kind, basis_act, act_number, act_date,
|
||
purpose, doc_url}]. Пустой список если cad_num=None / нет записей /
|
||
таблица не задеплоена.
|
||
"""
|
||
if not cad_num:
|
||
return []
|
||
try:
|
||
rows = db.execute(_LOOKUP_SQL, {"cad_num": cad_num}).mappings().all()
|
||
except (OperationalError, ProgrammingError) as exc:
|
||
# Таблица ещё не задеплоена / БД-ошибка — graceful degrade.
|
||
logger.warning("parcel_reservations: land_reservation недоступна, skip: %s", exc)
|
||
return []
|
||
|
||
return [
|
||
{
|
||
"reservation_kind": r["reservation_kind"],
|
||
"basis_act": r["basis_act"],
|
||
"act_number": r["act_number"],
|
||
"act_date": r["act_date"],
|
||
"purpose": r["purpose"],
|
||
"doc_url": r["doc_url"],
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
__all__ = ["parcel_reservations"]
|