feat(db): quarter_price_index FDW foreign table + monthly refresh (Refs #762) #797

Merged
bot-reviewer merged 1 commit from feat/762-quarter-index-refresh into main 2026-05-30 17:16:28 +00:00
5 changed files with 214 additions and 0 deletions

View file

@ -0,0 +1,85 @@
"""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

View file

@ -264,6 +264,16 @@ def build_beat_schedule() -> dict:
# "options": {"queue": "celery"},
# }
# Quarter price index refresh (#762) — monthly on the 5th at 05:00 MSK (02:00 UTC).
# Chains: mv_quarter_price_per_m2 → mv_quarter_price_index (in order, both CONCURRENTLY).
# Runs 1 h after refresh-ekb-districts-medians (04:00 MSK) so deals data is settled.
# Not in job_settings (technical infra task, no UI config needed).
schedule["refresh-quarter-price-index"] = {
"task": "tasks.refresh_quarter_price_index.refresh_quarter_price_index_chain",
"schedule": _parse_cron("0 2 5 * *"), # 02:00 UTC = 05:00 MSK, 5th of month
"options": {"queue": "celery"},
}
# NSPD quarter dump refresh — re-enabled 2026-05-17 после Sub-PR B (#260)
# переключения search_by_quarter на grid-walk. Foundation (#247) + integration
# (#260) теперь возвращают полноценные dumps (territorial_zones, ЗОУИТ, risk

View file

@ -0,0 +1,45 @@
"""Celery task: refresh mv_quarter_price_per_m2 → mv_quarter_price_index chain.
Scheduled via hardcoded beat entry in workers/beat_schedule.py:
'refresh-quarter-price-index' monthly on the 5th at 05:00 MSK (02:00 UTC).
Runs after 'refresh-ekb-districts-medians' (04:00 MSK same day) so that
deals data is settled before the price index is recomputed.
Issue: #762.
"""
from __future__ import annotations
import logging
from typing import Any
from app.core.db import SessionLocal
from app.services.site_finder.quarter_price_index_refresh import refresh_quarter_price_index
from app.workers.celery_app import celery_app
logger = logging.getLogger(__name__)
@celery_app.task(
bind=True,
name="tasks.refresh_quarter_price_index.refresh_quarter_price_index_chain",
max_retries=2,
)
def refresh_quarter_price_index_chain(self: Any) -> dict[str, Any]:
"""Refresh mv_quarter_price_per_m2 then mv_quarter_price_index in sequence.
Both MVs are refreshed CONCURRENTLY (non-blocking). Falls back to
non-concurrent if the MV is found unpopulated (edge case on first run).
Returns result dict for Celery task result store / logging.
"""
db = SessionLocal()
try:
count = refresh_quarter_price_index(db, concurrently=True)
logger.info("refresh_quarter_price_index_chain: completed, index rows=%d", count)
return {"status": "ok", "mv_quarter_price_index_rows": count}
except Exception as e:
logger.exception("refresh_quarter_price_index_chain failed: %s", e)
raise
finally:
db.close()

View file

@ -0,0 +1,16 @@
-- 99b_grant_quarter_price_index_fdw.sql
-- Issue #762 (#647-2) — grant SELECT on mv_quarter_price_index to the tradein
-- FDW reader role so the tradein foreign table (085_quarter_price_index_fdw.sql)
-- can read it. Mirrors 122_grant_rosreestr_to_fdw_reader.sql.
--
-- Ordering: '99b_' sorts AFTER '99a_quarter_price_index.sql' (which CREATEs the MV)
-- because 'a' < 'b'. The MV must exist before the GRANT. Files 100_122_ sort
-- BEFORE 99_ lexicographically ('1' < '9'), so they cannot grant on this object.
-- Idempotent: GRANT is repeatable; role tradein_fdw_reader created in
-- 100_tradein_fdw_role.sql (REVOKE ALL baseline there).
BEGIN;
GRANT SELECT ON public.mv_quarter_price_index TO tradein_fdw_reader;
COMMIT;

View file

@ -0,0 +1,58 @@
-- 085_quarter_price_index_fdw.sql
-- Issue #762 — Foreign table over MAIN DB's mv_quarter_price_index MV.
--
-- Context:
-- #760 (PR #792) created public.mv_quarter_price_index in the MAIN gendesign DB.
-- The estimator (#647-3) needs O(1) lookup by quarter_cad_number without a
-- cross-service HTTP round-trip. FDW foreign table lets tradein read the MV
-- directly via the already-established gendesign_remote server.
--
-- Reused infra:
-- SERVER gendesign_remote — created by 060_postgres_fdw_extension.sql.
-- Pattern: per-table FOREIGN TABLE (not IMPORT FOREIGN SCHEMA) — matches repo convention.
-- USER MAPPING is managed at backend startup from env (see 060_postgres_fdw_extension.sql).
--
-- Column mapping (verified against live MV via pg_attribute, 2026-05-30):
-- quarter_cad_number character varying(30) → varchar(30) / text (FDW accepts text for varchar)
-- price_index double precision → double precision
-- n_deals bigint → bigint
-- basis text → text
-- computed_at timestamp with time zone → timestamptz
--
-- Index note:
-- The remote MV has UNIQUE index mv_quarter_price_index_uq on quarter_cad_number.
-- FDW pushes equality predicates (WHERE quarter_cad_number = $1) down to MAIN,
-- so remote index gives O(1) lookup. No local index on the foreign table is needed
-- or possible (postgres_fdw does not support local indexes on foreign tables).
--
-- Idempotency:
-- DROP FOREIGN TABLE IF EXISTS → CREATE is safe to re-apply.
-- BEGIN/COMMIT per sql.md conventions.
--
-- Apply order:
-- After 060_postgres_fdw_extension.sql (server must exist).
-- After 99a_quarter_price_index.sql is applied in MAIN (MV must exist remotely).
-- Before estimator code referencing this table is deployed (#647-3).
BEGIN;
DROP FOREIGN TABLE IF EXISTS quarter_price_index;
CREATE FOREIGN TABLE quarter_price_index (
quarter_cad_number varchar(30) NOT NULL,
price_index double precision NOT NULL,
n_deals bigint NOT NULL,
basis text NOT NULL,
computed_at timestamp with time zone NOT NULL
)
SERVER gendesign_remote
OPTIONS (schema_name 'public', table_name 'mv_quarter_price_index');
COMMENT ON FOREIGN TABLE quarter_price_index IS
'FDW read-only view of gendesign.public.mv_quarter_price_index. '
'Per-cadastral-quarter price index normalised to EKB city median (=1.0). '
'Lookup by quarter_cad_number is O(1) via remote UNIQUE index on MAIN. '
'Refresh cadence: MAIN refreshes MV via Celery beat (see quarter_price_index_refresh task). '
'Issue: #762. Source MV: #760.';
COMMIT;