fix(velocity): SAVEPOINT per query (not db.rollback) — correct SQLAlchemy 2.0 pattern

Per PR #155 bot review — мой db.rollback() в catches был НЕПРАВИЛЬНЫЙ.

Session.rollback() = ENTIRE outer tx (не savepoint). Outer SAVEPOINT context
становится orphaned → InvalidRequestError на __exit__.

Correct fix: wrap каждый db.execute в свой with db.begin_nested():
- Failure → __exit__ propagates exception → SAVEPOINT rolls back
- Outer tx остаётся clean
- velocity returns None gracefully → caller продолжает

Все 3 SQL queries в velocity.py теперь SAVEPOINT-wrapped:
- compute_velocity: competitor_query
- compute_velocity: sales_rows query
- _get_ekb_median: median query

db.rollback() removed из всех 3 catches.

Refs: PR #155 bot review
This commit is contained in:
lekss361 2026-05-15 08:49:28 +03:00
parent c3cafefb1a
commit 37fd4b5a8e

View file

@ -74,7 +74,11 @@ def compute_velocity(
# obj_class в domrf_kn_objects заполнен слабо (много NULL); фильтруем # obj_class в domrf_kn_objects заполнен слабо (много NULL); фильтруем
# только если явно передан. # только если явно передан.
class_filter = "AND o.obj_class = :obj_class" if obj_class else "" class_filter = "AND o.obj_class = :obj_class" if obj_class else ""
# SAVEPOINT per query: failure rollbacks ТОЛЬКО savepoint, не outer tx.
# db.rollback() здесь НЕЛЬЗЯ — он orphan'ит outer SessionTransaction
# (см. PR #155 bot review — SQLAlchemy 2.0 begin_nested context cleanup).
try: try:
with db.begin_nested():
comp_rows = ( comp_rows = (
db.execute( db.execute(
text( text(
@ -126,13 +130,8 @@ def compute_velocity(
) )
except Exception: except Exception:
logger.exception("velocity: competitor query failed for wkt=%s", parcel_geom_wkt[:80]) logger.exception("velocity: competitor query failed for wkt=%s", parcel_geom_wkt[:80])
# КРИТИЧНО: rollback aborted transaction чтобы caller (analyze_parcel) # SAVEPOINT auto-rollbacks через __exit__ context manager.
# мог продолжить выполнять последующие queries. Иначе caller получает # Outer tx остаётся clean — caller продолжает работать без cascade.
# cascade InFailedSqlTransaction на следующей SQL execute.
try:
db.rollback()
except Exception:
pass
return None return None
if not comp_rows: if not comp_rows:
@ -154,6 +153,7 @@ def compute_velocity(
# area_sq = м² за месяц (primary). Если NULL — realised * 45 м² heuristic. # area_sq = м² за месяц (primary). Если NULL — realised * 45 м² heuristic.
# type = 'apartments' — только жильё. # type = 'apartments' — только жильё.
try: try:
with db.begin_nested():
sales_rows = ( sales_rows = (
db.execute( db.execute(
text( text(
@ -193,11 +193,7 @@ def compute_velocity(
) )
except Exception: except Exception:
logger.exception("velocity: sale_graph query failed for obj_ids=%s", obj_ids[:5]) logger.exception("velocity: sale_graph query failed for obj_ids=%s", obj_ids[:5])
# КРИТИЧНО: rollback aborted transaction (см. competitor query комментарий) # SAVEPOINT auto-rollback'нут — outer tx clean
try:
db.rollback()
except Exception:
pass
return None return None
if not sales_rows: if not sales_rows:
@ -278,6 +274,7 @@ def _get_ekb_median(db: Session, months_window: int = 6) -> float | None:
Fallback к _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH если нет данных в БД. Fallback к _EKB_MEDIAN_FALLBACK_SQM_PER_MONTH если нет данных в БД.
""" """
try: try:
with db.begin_nested():
row = ( row = (
db.execute( db.execute(
text( text(
@ -323,11 +320,7 @@ def _get_ekb_median(db: Session, months_window: int = 6) -> float | None:
) )
except Exception: except Exception:
logger.warning("velocity: ekb_median query failed, using fallback") logger.warning("velocity: ekb_median query failed, using fallback")
# КРИТИЧНО: rollback aborted transaction # SAVEPOINT auto-rollback'нут
try:
db.rollback()
except Exception:
pass
return None return None
if row and row["median"] is not None: if row and row["median"] is not None: