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 sys
|
||||
from collections.abc import Iterator
|
||||
from datetime import date, datetime
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
from typing import Any, BinaryIO
|
||||
|
||||
|
|
@ -597,26 +597,36 @@ def parse_lots_pf_stream(
|
|||
n_lots = 0
|
||||
n_history = 0
|
||||
batch: list[dict] = []
|
||||
batches_since_commit = 0
|
||||
|
||||
def _flush(batch: list[dict]) -> None:
|
||||
for params in batch:
|
||||
db.execute(_LOTS_UPSERT, params)
|
||||
db.execute(_LOTS_HISTORY_INSERT, {
|
||||
"objective_lot_id": params["objective_lot_id"],
|
||||
def _flush() -> None:
|
||||
"""Bulk-upsert текущего batch двумя executemany-вызовами."""
|
||||
nonlocal batch
|
||||
if not batch:
|
||||
return
|
||||
history_batch = [
|
||||
{
|
||||
"objective_lot_id": p["objective_lot_id"],
|
||||
"snapshot_date": snapshot_date,
|
||||
"status": params["status"],
|
||||
"is_sold": params["is_sold"],
|
||||
"price_calculated_total_rub": params["price_calculated_total_rub"],
|
||||
"price_per_m2_rub": params["price_per_m2_rub"],
|
||||
"price_offer_total_rub": params["price_offer_total_rub"],
|
||||
"price_delta_pct": params["price_delta_pct"],
|
||||
"area_pd": params["area_pd"],
|
||||
"contract_date": params["contract_date"],
|
||||
"registration_date": params["registration_date"],
|
||||
"contract_type": params["contract_type"],
|
||||
"bank_name": params["bank_name"],
|
||||
"status": p["status"],
|
||||
"is_sold": p["is_sold"],
|
||||
"price_calculated_total_rub": p["price_calculated_total_rub"],
|
||||
"price_per_m2_rub": p["price_per_m2_rub"],
|
||||
"price_offer_total_rub": p["price_offer_total_rub"],
|
||||
"price_delta_pct": p["price_delta_pct"],
|
||||
"area_pd": p["area_pd"],
|
||||
"contract_date": p["contract_date"],
|
||||
"registration_date": p["registration_date"],
|
||||
"contract_type": p["contract_type"],
|
||||
"bank_name": p["bank_name"],
|
||||
"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"):
|
||||
if not isinstance(row, dict):
|
||||
|
|
@ -625,20 +635,20 @@ def parse_lots_pf_stream(
|
|||
if not params["objective_lot_id"]:
|
||||
logger.warning("lots_pf_stream: пропуск row без Id")
|
||||
continue
|
||||
batch.append(params)
|
||||
n_lots += 1
|
||||
n_history += 1
|
||||
if not dry_run:
|
||||
batch.append(params)
|
||||
if len(batch) >= batch_size:
|
||||
_flush(batch)
|
||||
if len(batch) >= batch_size:
|
||||
_flush()
|
||||
batches_since_commit += 1
|
||||
if not dry_run and batches_since_commit >= 5:
|
||||
db.commit()
|
||||
batch.clear()
|
||||
logger.info("lots_pf_stream: committed batch, n_lots=%d so far", n_lots)
|
||||
batches_since_commit = 0
|
||||
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()
|
||||
batch.clear()
|
||||
|
||||
logger.info(
|
||||
"lots_pf_stream: done — n_lots=%d n_history=%d dry_run=%s",
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue