fix(pzz_loader): replace bare db.rollback() with SAVEPOINT per-row (#124)

Per code review audit (May 14): transaction correctness bug в
`sync_pzz_zones_to_db()` — inner `db.rollback()` в loop сбрасывал outer
transaction на per-row failures → counts inserted/updated НЕ matched DB state.

## Fix

Wrap UPSERT в loop через `db.begin_nested()` (SAVEPOINT):
- Failure одной row → rollback только её savepoint, outer tx alive
- Successfully inserted rows accumulate в outer tx → commit final
- Outer try/except добавлен для unexpected error → rollback + log + raise
- `finally: db.close()` preserved

Pattern consistent с `domrf_kn.py` (где SAVEPOINT уже работает).

## Preserved

- Function signature + return shape {fetched, inserted, updated, skipped} — unchanged
- Logging contracts
- skipped count теперь корректно включает DB-failure rows (consistent with prior behavior)

## Tests

- No test_pzz_loader.py — skipped
- ruff + AST passed

## Vault

`fixes/Bug_Pzz_Loader_Missing_Savepoint_May14.md` — created (status: resolved).

Co-authored-by: lekss361 <claudestars@proton.me>
This commit is contained in:
lekss361 2026-05-14 23:29:23 +03:00 committed by GitHub
parent 8dfab7613d
commit a3e7d21e3a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -80,6 +80,10 @@ def sync_pzz_zones_to_db() -> dict[str, int]:
"""Загрузить зоны из PKK6 и UPSERT в pzz_zones_ekb. """Загрузить зоны из PKK6 и UPSERT в pzz_zones_ekb.
Возвращает словарь {fetched, inserted, updated, skipped}. Возвращает словарь {fetched, inserted, updated, skipped}.
Каждый UPSERT обёрнут в SAVEPOINT (db.begin_nested), чтобы ошибка
одной записи не откатывала всю транзакцию и счётчики оставались
согласованными с тем, что реально записано в DB.
""" """
features = asyncio.run(fetch_pkk6_zones()) features = asyncio.run(fetch_pkk6_zones())
inserted = 0 inserted = 0
@ -105,41 +109,45 @@ def sync_pzz_zones_to_db() -> dict[str, int]:
) )
description = props.get("description") description = props.get("description")
try: try:
result = db.execute( with db.begin_nested(): # SAVEPOINT — откат только этой записи
text(""" result = db.execute(
INSERT INTO pzz_zones_ekb text("""
(rosreestr_id, zone_code, zone_name, description, raw_props, geom) INSERT INTO pzz_zones_ekb
VALUES ( (rosreestr_id, zone_code, zone_name, description, raw_props, geom)
:rid, :code, :nm, :desc, CAST(:props AS jsonb), VALUES (
ST_Multi(ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326)) :rid, :code, :nm, :desc, CAST(:props AS jsonb),
) ST_Multi(ST_SetSRID(ST_GeomFromGeoJSON(:g), 4326))
ON CONFLICT (rosreestr_id) DO UPDATE )
SET zone_code = EXCLUDED.zone_code, ON CONFLICT (rosreestr_id) DO UPDATE
zone_name = EXCLUDED.zone_name, SET zone_code = EXCLUDED.zone_code,
description = EXCLUDED.description, zone_name = EXCLUDED.zone_name,
raw_props = EXCLUDED.raw_props, description = EXCLUDED.description,
geom = EXCLUDED.geom, raw_props = EXCLUDED.raw_props,
fetched_at = NOW() geom = EXCLUDED.geom,
RETURNING (xmax = 0) AS is_insert fetched_at = NOW()
"""), RETURNING (xmax = 0) AS is_insert
{ """),
"rid": rosreestr_id, {
"code": zone_code, "rid": rosreestr_id,
"nm": zone_name, "code": zone_code,
"desc": description, "nm": zone_name,
"props": json.dumps(props, ensure_ascii=False), "desc": description,
"g": json.dumps(geom), "props": json.dumps(props, ensure_ascii=False),
}, "g": json.dumps(geom),
).scalar() },
).scalar()
if result: if result:
inserted += 1 inserted += 1
else: else:
updated += 1 updated += 1
except Exception as e: except Exception as e:
logger.warning("PZZ insert failed for rosreestr_id=%s: %s", rosreestr_id, e) logger.warning("PZZ insert failed for rosreestr_id=%s: %s", rosreestr_id, e)
db.rollback()
skipped += 1 skipped += 1
db.commit() db.commit()
except Exception as e:
db.rollback()
logger.exception("pzz_loader: unexpected error, outer tx rolled back: %s", e)
raise
finally: finally:
db.close() db.close()
return { return {