bot-backend
76baf8011d
#647-2 (P0, инфра — клиентскую цену НЕ меняет). Делает mv_quarter_price_index (#760) читаемым estimator'ом за O(1) + периодический refresh. - tradein 085_quarter_price_index_fdw.sql: FOREIGN TABLE quarter_price_index над main public.mv_quarter_price_index через существующий server gendesign_remote (паттерн 060_postgres_fdw_extension.sql, per-table). Колонки 1:1 с MV (varchar(30)/float8/bigint/text/timestamptz). DROP IF EXISTS идемпот. - main 99b_grant_quarter_price_index_fdw.sql: GRANT SELECT на MV роли tradein_fdw_reader (иначе foreign table падает; зеркало 122_grant_rosreestr). - main refresh: Celery beat (pg_cron нет). quarter_price_index_refresh.py — chained REFRESH mv_quarter_price_per_m2 → mv_quarter_price_index CONCURRENTLY; task refresh_quarter_price_index.py; beat-entry 'refresh-quarter-price-index' cron '0 2 5 * *' (05:00 MSK 5-го, после ekb-districts-medians). Lookup-индекс: MV уже несёт UNIQUE mv_quarter_price_index_uq (O(1) remote); foreign table не нуждается в локальном. Валидация (postgres-gendesign): MV 1972 строки, 0 NULL, UNIQUE indisunique=true, REFRESH CONCURRENTLY валиден. Разблокирует #647-3 (estimator integration, за гейтом #763).
85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
"""Refresh helper for mv_quarter_price_index (Issue #762).
|
|
|
|
mv_quarter_price_index depends on mv_quarter_price_per_m2, so both must be
|
|
refreshed in order:
|
|
1. mv_quarter_price_per_m2 (source MV — deals aggregation)
|
|
2. mv_quarter_price_index (derived MV — price index normalised to city median)
|
|
|
|
Scheduled via Celery beat hardcoded entry in workers/beat_schedule.py.
|
|
Cadence: monthly on the 5th at 05:00 MSK (after refresh_ekb_districts_medians
|
|
which runs at 04:00 MSK on the same day — deals data is already settled).
|
|
|
|
Usage example (manual, via psql-connected shell or admin endpoint):
|
|
from sqlalchemy.orm import Session
|
|
from app.services.site_finder.quarter_price_index_refresh import refresh_quarter_price_index
|
|
|
|
count = refresh_quarter_price_index(db)
|
|
# logs: "mv_quarter_price_index refreshed: 1972 rows"
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.exc import OperationalError
|
|
from sqlalchemy.orm import Session
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _refresh_mv(db: Session, mv_name: str, *, concurrently: bool) -> None:
|
|
"""Run REFRESH MATERIALIZED VIEW [CONCURRENTLY] <mv_name>.
|
|
|
|
Falls back to non-concurrent on the known "cannot refresh concurrently"
|
|
error (MV empty or no UNIQUE index — should not happen in prod, but
|
|
provides a safe recovery path for first-run edge cases).
|
|
"""
|
|
try:
|
|
if concurrently:
|
|
db.execute(text(f"REFRESH MATERIALIZED VIEW CONCURRENTLY {mv_name}"))
|
|
else:
|
|
db.execute(text(f"REFRESH MATERIALIZED VIEW {mv_name}"))
|
|
db.commit()
|
|
except OperationalError as e:
|
|
if concurrently and "cannot refresh materialized view" in str(e).lower():
|
|
logger.warning(
|
|
"%s: CONCURRENTLY failed (MV likely not populated), "
|
|
"falling back to non-concurrent refresh",
|
|
mv_name,
|
|
)
|
|
db.rollback()
|
|
db.execute(text(f"REFRESH MATERIALIZED VIEW {mv_name}"))
|
|
db.commit()
|
|
else:
|
|
raise
|
|
|
|
|
|
def refresh_quarter_price_index(db: Session, *, concurrently: bool = True) -> int:
|
|
"""Refresh mv_quarter_price_per_m2 then mv_quarter_price_index (in order).
|
|
|
|
mv_quarter_price_index reads from mv_quarter_price_per_m2, so the source MV
|
|
must be refreshed first. Both refreshes happen in the same call.
|
|
|
|
Args:
|
|
db: SQLAlchemy Session (sync).
|
|
concurrently: When True, uses REFRESH CONCURRENTLY for both MVs —
|
|
non-blocking (readers continue). Requires the respective UNIQUE
|
|
indexes (mv_quarter_price_pk on source, mv_quarter_price_index_uq
|
|
on derived) and both MVs to be already populated.
|
|
Pass False only for first populate or after MV recreation.
|
|
|
|
Returns:
|
|
Row count of mv_quarter_price_index after refresh (for observability).
|
|
"""
|
|
# Step 1: refresh source MV (deals aggregation layer)
|
|
_refresh_mv(db, "mv_quarter_price_per_m2", concurrently=concurrently)
|
|
logger.info("mv_quarter_price_per_m2 refreshed (chain step 1/2)")
|
|
|
|
# Step 2: refresh derived MV (price index, reads from source)
|
|
_refresh_mv(db, "mv_quarter_price_index", concurrently=concurrently)
|
|
|
|
row = db.execute(text("SELECT COUNT(*) FROM mv_quarter_price_index")).first()
|
|
count = int(row[0]) if row else 0
|
|
logger.info("mv_quarter_price_index refreshed: %d rows (chain step 2/2)", count)
|
|
return count
|