fix(objective parser): restore bulk executemany in stream _flush
Reviewer flagged perf regression vs PR #315 — _flush loop'ил per-row через db.execute(stmt, dict) вместо bulk db.execute(stmt, list). Для 324k лотов это 648k round-trips × 2 sync/неделю. Восстановлен SQLAlchemy executemany через list[dict] + commit cadence 5 batch (2500 строк) как в #315. Также убран неиспользуемый импорт datetime (ruff F401). Per #318 (comment)
This commit is contained in:
parent
8d0e67d03b
commit
c8ac1fa732
1 changed files with 37 additions and 27 deletions
|
|
@ -41,7 +41,7 @@ import logging
|
||||||
import re
|
import re
|
||||||
import sys
|
import sys
|
||||||
from collections.abc import Iterator
|
from collections.abc import Iterator
|
||||||
from datetime import date, datetime
|
from datetime import date
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, BinaryIO
|
from typing import Any, BinaryIO
|
||||||
|
|
||||||
|
|
@ -597,26 +597,36 @@ def parse_lots_pf_stream(
|
||||||
n_lots = 0
|
n_lots = 0
|
||||||
n_history = 0
|
n_history = 0
|
||||||
batch: list[dict] = []
|
batch: list[dict] = []
|
||||||
|
batches_since_commit = 0
|
||||||
|
|
||||||
def _flush(batch: list[dict]) -> None:
|
def _flush() -> None:
|
||||||
for params in batch:
|
"""Bulk-upsert текущего batch двумя executemany-вызовами."""
|
||||||
db.execute(_LOTS_UPSERT, params)
|
nonlocal batch
|
||||||
db.execute(_LOTS_HISTORY_INSERT, {
|
if not batch:
|
||||||
"objective_lot_id": params["objective_lot_id"],
|
return
|
||||||
|
history_batch = [
|
||||||
|
{
|
||||||
|
"objective_lot_id": p["objective_lot_id"],
|
||||||
"snapshot_date": snapshot_date,
|
"snapshot_date": snapshot_date,
|
||||||
"status": params["status"],
|
"status": p["status"],
|
||||||
"is_sold": params["is_sold"],
|
"is_sold": p["is_sold"],
|
||||||
"price_calculated_total_rub": params["price_calculated_total_rub"],
|
"price_calculated_total_rub": p["price_calculated_total_rub"],
|
||||||
"price_per_m2_rub": params["price_per_m2_rub"],
|
"price_per_m2_rub": p["price_per_m2_rub"],
|
||||||
"price_offer_total_rub": params["price_offer_total_rub"],
|
"price_offer_total_rub": p["price_offer_total_rub"],
|
||||||
"price_delta_pct": params["price_delta_pct"],
|
"price_delta_pct": p["price_delta_pct"],
|
||||||
"area_pd": params["area_pd"],
|
"area_pd": p["area_pd"],
|
||||||
"contract_date": params["contract_date"],
|
"contract_date": p["contract_date"],
|
||||||
"registration_date": params["registration_date"],
|
"registration_date": p["registration_date"],
|
||||||
"contract_type": params["contract_type"],
|
"contract_type": p["contract_type"],
|
||||||
"bank_name": params["bank_name"],
|
"bank_name": p["bank_name"],
|
||||||
"raw_id": raw_id,
|
"raw_id": raw_id,
|
||||||
})
|
}
|
||||||
|
for p in batch
|
||||||
|
]
|
||||||
|
if not dry_run:
|
||||||
|
db.execute(_LOTS_UPSERT, batch)
|
||||||
|
db.execute(_LOTS_HISTORY_INSERT, history_batch)
|
||||||
|
batch = []
|
||||||
|
|
||||||
for row in ijson.items(stream, "result.item"):
|
for row in ijson.items(stream, "result.item"):
|
||||||
if not isinstance(row, dict):
|
if not isinstance(row, dict):
|
||||||
|
|
@ -625,20 +635,20 @@ def parse_lots_pf_stream(
|
||||||
if not params["objective_lot_id"]:
|
if not params["objective_lot_id"]:
|
||||||
logger.warning("lots_pf_stream: пропуск row без Id")
|
logger.warning("lots_pf_stream: пропуск row без Id")
|
||||||
continue
|
continue
|
||||||
|
batch.append(params)
|
||||||
n_lots += 1
|
n_lots += 1
|
||||||
n_history += 1
|
n_history += 1
|
||||||
if not dry_run:
|
if len(batch) >= batch_size:
|
||||||
batch.append(params)
|
_flush()
|
||||||
if len(batch) >= batch_size:
|
batches_since_commit += 1
|
||||||
_flush(batch)
|
if not dry_run and batches_since_commit >= 5:
|
||||||
db.commit()
|
db.commit()
|
||||||
batch.clear()
|
batches_since_commit = 0
|
||||||
logger.info("lots_pf_stream: committed batch, n_lots=%d so far", n_lots)
|
logger.info("lots_pf_stream: committed 5 batches, n_lots=%d so far", n_lots)
|
||||||
|
|
||||||
if not dry_run and batch:
|
_flush() # финальный неполный batch
|
||||||
_flush(batch)
|
if not dry_run and batches_since_commit > 0:
|
||||||
db.commit()
|
db.commit()
|
||||||
batch.clear()
|
|
||||||
|
|
||||||
logger.info(
|
logger.info(
|
||||||
"lots_pf_stream: done — n_lots=%d n_history=%d dry_run=%s",
|
"lots_pf_stream: done — n_lots=%d n_history=%d dry_run=%s",
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue