Persist each successful POST /analyze into analysis_runs (migration 127): best-effort SAVEPOINT-wrapped INSERT in a thin analysis_runs repository, then explicit db.commit() (get_db has no commit-on-success; SAVEPOINT RELEASE alone does not persist — without the commit the row rolls back on request teardown, a silent no-op caught in code review). A persist failure never breaks the response or poisons the session. Result serialized via jsonable_encoder→ json.dumps (handles embedded Pydantic models/dates); confidence/status normalized to satisfy the 127 CHECK constraints (unknown→NULL/'complete'). Populate by-bbox last_analysis_date from v_analysis_runs_latest via a single batch query (no N+1), replacing the #307 placeholder. The read is best-effort wrapped too — a view-missing deploy window or future drift must not 500 the map (falls back to last_analysis_date=None). Additive only — analyze response shape unchanged. Tests: by-bbox suite patches latest_run_dates (mock-db returns same rows for any query) + new test asserts ISO last_analysis_date from a latest run. Analyze mock-suite unaffected (graceful side_effect overflow + best-effort persist absorb the extra INSERT). Incidental: ruff-format fixed one pre-existing f-string spacing (line ~2391) the format hook flags on touch. Closes #994. Refs #961.
195 lines
9.5 KiB
Python
195 lines
9.5 KiB
Python
"""Data-access для analysis_runs (#994, 961-C3, ТЗ §22, EPIC 17 «GG-форсайт»).
|
||
|
||
Тонкий repository поверх таблицы `analysis_runs` + view `v_analysis_runs_latest`
|
||
(миграция 127 / #993). Держит endpoint `analyze_parcel` тонким: persist одного
|
||
завершённого рана + чтение «даты последнего анализа» (powers by-bbox карту).
|
||
|
||
Контракт миграции 127 (см. data/sql/127_analysis_runs.sql):
|
||
- result/params/segment — JSONB; пишем через CAST(:x AS jsonb) + json.dumps
|
||
(psycopg v3: НИКОГДА :x::type — backend.md). result может содержать Pydantic-
|
||
модели (RiskZone/OpportunityParcel/RedLine) → прогоняем через jsonable_encoder
|
||
перед json.dumps, иначе TypeError на сериализации.
|
||
- CHECK'и: status ∈ {complete,partial,error}; confidence ∈ {high,medium,low} ИЛИ
|
||
NULL. Чтобы НИКОГДА не словить IntegrityError на write — валидируем обе колонки
|
||
локально (`_normalize_confidence` / `_normalize_status`): неизвестное → NULL/дефолт,
|
||
не сырое значение. Плохой ран не должен превращать успешный analyze в 500.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from fastapi.encoders import jsonable_encoder
|
||
from sqlalchemy import Row, text
|
||
from sqlalchemy.orm import Session
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Версия схемы рана, когда result — это dict эндпоинта analyze (НЕ SiteFinderReport).
|
||
# SiteFinderReport.as_dict() несёт собственный _SCHEMA_VERSION ("1.0"); для inline-
|
||
# результата analyze, у которого своей версии формы нет, фиксируем ОТДЕЛЬНУЮ строку
|
||
# с префиксом, чтобы re-open мог различить два источника (#992 формализует контракт).
|
||
ANALYZE_SCHEMA_VERSION = "analyze-1.0"
|
||
|
||
# CHECK-зеркала миграции 127 — единственный источник допустимых значений на write.
|
||
_ALLOWED_CONFIDENCE: frozenset[str] = frozenset({"high", "medium", "low"})
|
||
_ALLOWED_STATUS: frozenset[str] = frozenset({"complete", "partial", "error"})
|
||
_DEFAULT_STATUS = "complete"
|
||
|
||
|
||
def _normalize_confidence(confidence: str | None) -> str | None:
|
||
"""Привести confidence к CHECK-допустимому high|medium|low или None.
|
||
|
||
Пустая строка / неизвестное значение → None (а НЕ '' — это нарушило бы CHECK).
|
||
"""
|
||
if confidence is None:
|
||
return None
|
||
value = confidence.strip().lower()
|
||
return value if value in _ALLOWED_CONFIDENCE else None
|
||
|
||
|
||
def _normalize_status(status: str | None) -> str:
|
||
"""Привести status к CHECK-допустимому complete|partial|error (дефолт complete)."""
|
||
if status is None:
|
||
return _DEFAULT_STATUS
|
||
value = status.strip().lower()
|
||
return value if value in _ALLOWED_STATUS else _DEFAULT_STATUS
|
||
|
||
|
||
def _jsonb_param(value: Any) -> str:
|
||
"""Сериализовать произвольный (возможно Pydantic-содержащий) объект в JSON-строку.
|
||
|
||
jsonable_encoder разворачивает Pydantic-модели / даты / Enum в JSON-native типы;
|
||
json.dumps(..., ensure_ascii=False) — кириллица как есть (зеркало pzz_loader).
|
||
"""
|
||
return json.dumps(jsonable_encoder(value), ensure_ascii=False)
|
||
|
||
|
||
def persist_analysis_run(
|
||
db: Session,
|
||
*,
|
||
cad_num: str,
|
||
result: dict[str, Any],
|
||
params: dict[str, Any],
|
||
district: str | None,
|
||
confidence: str | None,
|
||
status: str,
|
||
schema_version: str,
|
||
created_by: str | None,
|
||
segment: dict[str, Any] | None = None,
|
||
) -> int | None:
|
||
"""Сохранить ОДИН завершённый ран анализа в analysis_runs (#994).
|
||
|
||
Best-effort: обёрнут в SAVEPOINT (db.begin_nested, backend.md) + try/except —
|
||
провал write НЕ ломает ответ analyze и НЕ отравляет outer-транзакцию (откатывает
|
||
только свой savepoint, прошлые reads сессии остаются валидны). На успехе —
|
||
возвращает id новой строки, на провале — None (+ logger.warning/exception).
|
||
|
||
confidence/status нормализуются под CHECK'и миграции 127 ДО write — IntegrityError
|
||
на enum'ах структурно невозможен. advisory всегда TRUE (форсайт-стек советующий).
|
||
"""
|
||
norm_confidence = _normalize_confidence(confidence)
|
||
norm_status = _normalize_status(status)
|
||
try:
|
||
with db.begin_nested():
|
||
row = (
|
||
db.execute(
|
||
text("""
|
||
INSERT INTO analysis_runs (
|
||
cad_num, district, segment, params, result,
|
||
schema_version, advisory, confidence, status, created_by
|
||
)
|
||
VALUES (
|
||
CAST(:cad_num AS text),
|
||
CAST(:district AS text),
|
||
CAST(:segment AS jsonb),
|
||
CAST(:params AS jsonb),
|
||
CAST(:result AS jsonb),
|
||
CAST(:schema_version AS text),
|
||
TRUE,
|
||
CAST(:confidence AS text),
|
||
CAST(:status AS text),
|
||
CAST(:created_by AS text)
|
||
)
|
||
RETURNING id
|
||
"""),
|
||
{
|
||
"cad_num": cad_num,
|
||
"district": district,
|
||
"segment": _jsonb_param(segment) if segment is not None else None,
|
||
"params": _jsonb_param(params),
|
||
"result": _jsonb_param(result),
|
||
"schema_version": schema_version,
|
||
"confidence": norm_confidence,
|
||
"status": norm_status,
|
||
"created_by": created_by,
|
||
},
|
||
)
|
||
.mappings()
|
||
.first()
|
||
)
|
||
new_id = int(row["id"]) if row else None
|
||
# SAVEPOINT RELEASE (выход из begin_nested) НЕ персистит — лишь сливает вложенную
|
||
# tx в outer. get_db() (core/db.py) коммита-на-успехе НЕ делает (yield→finally:close
|
||
# = ROLLBACK), а analyze — единственный write на success-path. Без явного commit
|
||
# строка откатится на teardown (silent no-op). Коммитим ВНУТРИ try → провал commit
|
||
# тоже best-effort. Паттерн как в trade_in.py / photos.py (explicit commit на write).
|
||
db.commit()
|
||
logger.info(
|
||
"analysis_run persisted: cad=%s id=%s status=%s confidence=%s",
|
||
cad_num,
|
||
new_id,
|
||
norm_status,
|
||
norm_confidence,
|
||
)
|
||
return new_id
|
||
except Exception:
|
||
# Persist — best-effort: ран уже посчитан, ответ отдаём в любом случае.
|
||
# exception (не warning) — нужен traceback, write-провалы должны быть видны.
|
||
logger.exception("analysis_run persist failed for cad=%s (response unaffected)", cad_num)
|
||
return None
|
||
|
||
|
||
def latest_run_dates(db: Session, cad_nums: list[str]) -> dict[str, datetime]:
|
||
"""Дата последнего анализа на участок для набора cad_num (batch, НЕ N+1).
|
||
|
||
Один запрос к v_analysis_runs_latest (canonical latest-per-parcel из 127):
|
||
`WHERE cad_num = ANY(CAST(:cads AS text[]))`. Powers last_analysis_date в by-bbox.
|
||
Участки без ранов в результат не попадают (caller трактует отсутствие как None).
|
||
Пустой вход → пустой dict (без обращения к БД).
|
||
"""
|
||
if not cad_nums:
|
||
return {}
|
||
rows = (
|
||
db.execute(
|
||
text("""
|
||
SELECT cad_num, created_at
|
||
FROM v_analysis_runs_latest
|
||
WHERE cad_num = ANY(CAST(:cads AS text[]))
|
||
"""),
|
||
{"cads": cad_nums},
|
||
)
|
||
.mappings()
|
||
.all()
|
||
)
|
||
return {row["cad_num"]: row["created_at"] for row in rows}
|
||
|
||
|
||
def latest_run_for(db: Session, cad_num: str) -> Row[Any] | None:
|
||
"""Последний ран на участок целиком (для re-open / «текущий анализ участка»).
|
||
|
||
Читает из v_analysis_runs_latest (DISTINCT ON cad_num, max created_at). None если
|
||
участок ещё не анализировался. Возвращает Row (id/result/params/.../created_at).
|
||
"""
|
||
return db.execute(
|
||
text("""
|
||
SELECT id, cad_num, district, segment, params, result,
|
||
schema_version, advisory, confidence, status, created_by, created_at
|
||
FROM v_analysis_runs_latest
|
||
WHERE cad_num = CAST(:cad_num AS text)
|
||
"""),
|
||
{"cad_num": cad_num},
|
||
).first()
|