diff --git a/backend/app/services/site_finder/ird_overlay_lookup.py b/backend/app/services/site_finder/ird_overlay_lookup.py index 33cc9b76..3d7681e8 100644 --- a/backend/app/services/site_finder/ird_overlay_lookup.py +++ b/backend/app/services/site_finder/ird_overlay_lookup.py @@ -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": {}} diff --git a/backend/app/services/site_finder/quarter_dump_lookup.py b/backend/app/services/site_finder/quarter_dump_lookup.py index 6dabac98..f066d9c7 100644 --- a/backend/app/services/site_finder/quarter_dump_lookup.py +++ b/backend/app/services/site_finder/quarter_dump_lookup.py @@ -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, diff --git a/backend/tests/services/test_ird_overlay_lookup.py b/backend/tests/services/test_ird_overlay_lookup.py index e5a0a945..3e9898c6 100644 --- a/backend/tests/services/test_ird_overlay_lookup.py +++ b/backend/tests/services/test_ird_overlay_lookup.py @@ -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