fix(site-finder): расширить except в parcel_ird_overlaps до DataError (#1095)

parcel_ird_overlaps ловил только (OperationalError, ProgrammingError).
При D9b-wiring в analyze malformed WKT в ST_GeomFromText давал PostGIS
ERROR → SQLAlchemy DataError, который пробивал try/except → analyze
падал вместо graceful-degrade.

- ird_overlay_lookup: + DataError в imports и в except tuple.
- quarter_dump_lookup: симметричное расширение в _get_engineering_*.
- test_ird_overlay_lookup: parametrized test_graceful_when_db_fails
  покрывает 3 класса (Operational/Programming + DataError для WKT).

Closes #1095
This commit is contained in:
Light1YT 2026-06-13 14:23:32 +05:00
parent d11c8d47e0
commit 2e428405d2
3 changed files with 42 additions and 11 deletions

View file

@ -19,7 +19,7 @@ import logging
from typing import Any
from sqlalchemy import text
from sqlalchemy.exc import OperationalError, ProgrammingError
from sqlalchemy.exc import DataError, OperationalError, ProgrammingError
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
@ -59,8 +59,12 @@ def parcel_ird_overlaps(db: Session, parcel_wkt: str | None) -> dict[str, Any]:
return {"ird_overlaps": [], "ird_by_kind": {}}
try:
rows = db.execute(_IRD_OVERLAP_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
except (OperationalError, ProgrammingError) as exc:
# Таблица ещё не задеплоена / БД-ошибка — graceful degrade.
except (OperationalError, ProgrammingError, DataError) as exc:
# OperationalError — БД недоступна / connection drop.
# ProgrammingError — ird_overlays ещё не задеплоена (UndefinedTable, pre-migration).
# DataError — malformed WKT в ST_GeomFromText (PostGIS ERROR пробрасывает SQLAlchemy
# как DataError/InternalError). parcel_wkt системно-генерируемый, но при D9b-wiring
# в analyze graceful-degrade — единственная корректная стратегия (analyze не падает).
logger.warning("parcel_ird_overlaps: ird_overlays недоступна, skip: %s", exc)
return {"ird_overlaps": [], "ird_by_kind": {}}

View file

@ -20,7 +20,7 @@ from datetime import UTC, datetime, timedelta
from typing import Any
from sqlalchemy import text
from sqlalchemy.exc import OperationalError, ProgrammingError
from sqlalchemy.exc import DataError, OperationalError, ProgrammingError
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
@ -1066,7 +1066,8 @@ def _get_engineering_structures_by_boundary(
),
{"q": quarter, "wkt": parcel_wkt, "radius_m": radius_m},
).fetchall()
except (OperationalError, ProgrammingError) as e:
except (OperationalError, ProgrammingError, DataError) as e:
# DataError — malformed WKT в ST_GeomFromText / ST_GeomFromGeoJSON (PostGIS ERROR).
logger.warning(
"engineering_structures query failed for quarter=%s: %s",
quarter,
@ -1139,7 +1140,8 @@ def _get_zouit_engineering_overlaps(
),
{"q": quarter, "wkt": parcel_wkt},
).fetchall()
except (OperationalError, ProgrammingError) as e:
except (OperationalError, ProgrammingError, DataError) as e:
# DataError — malformed WKT в ST_GeomFromText / ST_GeomFromGeoJSON (PostGIS ERROR).
logger.warning(
"zouit_engineering query failed for quarter=%s: %s",
quarter,

View file

@ -4,7 +4,8 @@ from __future__ import annotations
from typing import Any
from sqlalchemy.exc import OperationalError
import pytest
from sqlalchemy.exc import DataError, OperationalError, ProgrammingError
from app.services.site_finder.ird_overlay_lookup import parcel_ird_overlaps
@ -102,8 +103,32 @@ def test_empty_when_no_overlap() -> None:
assert res == {"ird_overlaps": [], "ird_by_kind": {}}
def test_graceful_when_table_missing() -> None:
"""ird_overlays ещё не задеплоена → пусто, без падения analyze."""
db = _FakeDB(raise_exc=OperationalError("stmt", {}, Exception("no such table")))
@pytest.mark.parametrize(
("exc", "case"),
[
(
OperationalError("stmt", {}, Exception("no such table")),
"operational-error: connection drop / БД недоступна",
),
(
ProgrammingError("stmt", {}, Exception("relation \"ird_overlays\" does not exist")),
"programming-error: UndefinedTable — pre-migration / D9b не задеплоен",
),
(
DataError("stmt", {}, Exception("parse error - invalid geometry")),
"data-error: malformed WKT → PostGIS ST_GeomFromText ERROR",
),
],
)
def test_graceful_when_db_fails(exc: Exception, case: str) -> None:
"""ird_overlays недоступна / malformed WKT → пусто, analyze не падает.
Покрывает:
- OperationalError БД недоступна.
- ProgrammingError таблица ещё не задеплоена (реальный класс при UndefinedTable).
- DataError malformed parcel_wkt (PostGIS ST_GeomFromText бросает ERROR,
SQLAlchemy конвертит в DataError).
"""
db = _FakeDB(raise_exc=exc)
res = parcel_ird_overlaps(db, _WKT)
assert res == {"ird_overlaps": [], "ird_by_kind": {}}
assert res == {"ird_overlaps": [], "ird_by_kind": {}}, case