lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

fix(site-finder): cascade InFailedSqlTransaction — velocity internal rollback #155

Merged
lekss361 merged 2 commits from fix/velocity-internal-rollback-cascade into main 2026-05-15 05:52:39 +00:00
Conversation 3 Commits 2 Files changed 2 +103 -91
2 changed files with 103 additions and 91 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

116
backend/app/api/v1/parcels.py
View file

@ -1658,33 +1658,34 @@ def analyze_parcel(
"lon": centroid_lon, "lon": centroid_lon,
} }
try: try:
zoning_row = ( with db.begin_nested():
db.execute( zoning_row = (
text(""" db.execute(
SELECT zone_code, zone_name, description, rosreestr_id text("""
FROM pzz_zones_ekb SELECT zone_code, zone_name, description, rosreestr_id
WHERE ST_Within( FROM pzz_zones_ekb
ST_Centroid(ST_GeomFromText(:wkt, 4326)), WHERE ST_Within(
geom ST_Centroid(ST_GeomFromText(:wkt, 4326)),
) geom
LIMIT 1 )
"""), LIMIT 1
{"wkt": geom_wkt}, """),
) {"wkt": geom_wkt},
.mappings() )
.first() .mappings()
) .first()
if zoning_row:
zoning.update(
{
"zone_code": zoning_row["zone_code"],
"zone_name": zoning_row["zone_name"],
"description": zoning_row["description"],
"rosreestr_id": zoning_row["rosreestr_id"],
"data_available": True,
"source": "rosreestr-pkk6-cached",
}
) )
if zoning_row:
zoning.update(
{
"zone_code": zoning_row["zone_code"],
"zone_name": zoning_row["zone_name"],
"description": zoning_row["description"],
"rosreestr_id": zoning_row["rosreestr_id"],
"data_available": True,
"source": "rosreestr-pkk6-cached",
}
)
except Exception as e: except Exception as e:
logger.warning("zoning query failed for %s: %s", cad_num, e) logger.warning("zoning query failed for %s: %s", cad_num, e)
@ -1692,9 +1693,10 @@ def analyze_parcel(
success_recommendation: dict[str, Any] | None = None success_recommendation: dict[str, Any] | None = None
if district_row: if district_row:
try: try:
success_rows = ( with db.begin_nested():
db.execute( success_rows = (
text(""" db.execute(
text("""
SELECT bucket, success_score, n_deals, avg_price_per_m2, avg_area_m2, SELECT bucket, success_score, n_deals, avg_price_per_m2, avg_area_m2,
velocity_z, price_z, area_z velocity_z, price_z, area_z
FROM v_bucket_success_score FROM v_bucket_success_score
@ -1702,35 +1704,35 @@ def analyze_parcel(
ORDER BY success_score DESC ORDER BY success_score DESC
LIMIT 5 LIMIT 5
"""), """),
{"dn": district_row["district_name"]}, {"dn": district_row["district_name"]},
)
.mappings()
.all()
) )
.mappings() if success_rows:
.all() success_recommendation = {
) "district": district_row["district_name"],
if success_rows: "ranking": [
success_recommendation = { {
"district": district_row["district_name"], "bucket": r["bucket"],
"ranking": [ "success_score": round(float(r["success_score"]), 2),
{ "n_deals": int(r["n_deals"]),
"bucket": r["bucket"], "avg_price_per_m2": (
"success_score": round(float(r["success_score"]), 2), int(r["avg_price_per_m2"]) if r["avg_price_per_m2"] else None
"n_deals": int(r["n_deals"]), ),
"avg_price_per_m2": ( "avg_area_m2": (
int(r["avg_price_per_m2"]) if r["avg_price_per_m2"] else None round(float(r["avg_area_m2"]), 1) if r["avg_area_m2"] else None
), ),
"avg_area_m2": ( }
round(float(r["avg_area_m2"]), 1) if r["avg_area_m2"] else None for r in success_rows
), ],
} "top_bucket": dict(success_rows[0]) if success_rows else None,
for r in success_rows "note": (
], "Топ комнатность по 'успешности' = z-scores: velocity×0.5 + price×0.3 "
"top_bucket": dict(success_rows[0]) if success_rows else None, "- area×0.2. Min 30 сделок в группе за 24 мес. "
"note": ( "Используй для квартирографии проекта."
"Топ комнатность по 'успешности' = z-scores: velocity×0.5 + price×0.3 " ),
"- area×0.2. Min 30 сделок в группе за 24 мес. " }
"Используй для квартирографии проекта."
),
}
except Exception as e: except Exception as e:
logger.warning("success_recommendation query failed for %s: %s", cad_num, e) logger.warning("success_recommendation query failed for %s: %s", cad_num, e)
success_recommendation = None success_recommendation = None

78
backend/app/services/site_finder/velocity.py
View file

@ -74,11 +74,15 @@ 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:
comp_rows = ( with db.begin_nested():
db.execute( comp_rows = (
text( db.execute(
f""" text(
f"""
WITH latest_obj AS ( WITH latest_obj AS (
SELECT DISTINCT ON (obj_id) SELECT DISTINCT ON (obj_id)
obj_id, obj_id,
@ -114,18 +118,20 @@ def compute_velocity(
ORDER BY distance_m ASC ORDER BY distance_m ASC
LIMIT 200 LIMIT 200
""" """
), ),
{ {
"parcel_wkt": parcel_geom_wkt, "parcel_wkt": parcel_geom_wkt,
"radius_m": radius_km * 1000.0, "radius_m": radius_km * 1000.0,
"obj_class": obj_class, "obj_class": obj_class,
}, },
)
.mappings()
.all()
) )
.mappings()
.all()
)
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])
# SAVEPOINT auto-rollbacks через __exit__ context manager.
# Outer tx остаётся clean — caller продолжает работать без cascade.
return None return None
if not comp_rows: if not comp_rows:
@ -147,10 +153,11 @@ def compute_velocity(
# area_sq = м² за месяц (primary). Если NULL — realised * 45 м² heuristic. # area_sq = м² за месяц (primary). Если NULL — realised * 45 м² heuristic.
# type = 'apartments' — только жильё. # type = 'apartments' — только жильё.
try: try:
sales_rows = ( with db.begin_nested():
db.execute( sales_rows = (
text( db.execute(
""" text(
"""
WITH latest_sg AS ( WITH latest_sg AS (
SELECT DISTINCT ON (obj_id, report_month) SELECT DISTINCT ON (obj_id, report_month)
obj_id, obj_id,
@ -175,17 +182,18 @@ def compute_velocity(
WHERE area_sq > 0 OR realised > 0 WHERE area_sq > 0 OR realised > 0
GROUP BY obj_id GROUP BY obj_id
""" """
), ),
{ {
"obj_ids": obj_ids, "obj_ids": obj_ids,
"window_interval": f"{months_window} months", "window_interval": f"{months_window} months",
}, },
)
.mappings()
.all()
) )
.mappings()
.all()
)
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])
# SAVEPOINT auto-rollback'нут — outer tx clean
return None return None
if not sales_rows: if not sales_rows:
@ -266,10 +274,11 @@ 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:
row = ( with db.begin_nested():
db.execute( row = (
text( db.execute(
""" text(
"""
WITH latest_sg AS ( WITH latest_sg AS (
SELECT DISTINCT ON (obj_id, report_month) SELECT DISTINCT ON (obj_id, report_month)
obj_id, obj_id,
@ -303,14 +312,15 @@ def _get_ekb_median(db: Session, months_window: int = 6) -> float | None:
SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY velocity) AS median SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY velocity) AS median
FROM per_obj_velocity FROM per_obj_velocity
""" """
), ),
{"window_interval": f"{months_window} months"}, {"window_interval": f"{months_window} months"},
)
.mappings()
.first()
) )
.mappings()
.first()
)
except Exception: except Exception:
logger.warning("velocity: ekb_median query failed, using fallback") logger.warning("velocity: ekb_median query failed, using fallback")
# SAVEPOINT auto-rollback'нут
return None return None
if row and row["median"] is not None: if row and row["median"] is not None: