From 87a5de0caea6258e82842f2750db7e4dd3e3b8f8 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 30 May 2026 17:16:28 +0000 Subject: [PATCH] feat(db): quarter_price_index FDW foreign table + monthly refresh (Refs #762) (#797) Co-authored-by: bot-backend Co-committed-by: bot-backend --- .../quarter_price_index_refresh.py | 85 +++++++++++++++++++ backend/app/workers/beat_schedule.py | 10 +++ .../tasks/refresh_quarter_price_index.py | 45 ++++++++++ .../sql/99b_grant_quarter_price_index_fdw.sql | 16 ++++ .../data/sql/085_quarter_price_index_fdw.sql | 58 +++++++++++++ 5 files changed, 214 insertions(+) create mode 100644 backend/app/services/site_finder/quarter_price_index_refresh.py create mode 100644 backend/app/workers/tasks/refresh_quarter_price_index.py create mode 100644 data/sql/99b_grant_quarter_price_index_fdw.sql create mode 100644 tradein-mvp/backend/data/sql/085_quarter_price_index_fdw.sql diff --git a/backend/app/services/site_finder/quarter_price_index_refresh.py b/backend/app/services/site_finder/quarter_price_index_refresh.py new file mode 100644 index 00000000..56e7a558 --- /dev/null +++ b/backend/app/services/site_finder/quarter_price_index_refresh.py @@ -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] . + + 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 diff --git a/backend/app/workers/beat_schedule.py b/backend/app/workers/beat_schedule.py index 694cde42..4b9afd9e 100644 --- a/backend/app/workers/beat_schedule.py +++ b/backend/app/workers/beat_schedule.py @@ -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 diff --git a/backend/app/workers/tasks/refresh_quarter_price_index.py b/backend/app/workers/tasks/refresh_quarter_price_index.py new file mode 100644 index 00000000..803e3457 --- /dev/null +++ b/backend/app/workers/tasks/refresh_quarter_price_index.py @@ -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() diff --git a/data/sql/99b_grant_quarter_price_index_fdw.sql b/data/sql/99b_grant_quarter_price_index_fdw.sql new file mode 100644 index 00000000..b8feec96 --- /dev/null +++ b/data/sql/99b_grant_quarter_price_index_fdw.sql @@ -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; diff --git a/tradein-mvp/backend/data/sql/085_quarter_price_index_fdw.sql b/tradein-mvp/backend/data/sql/085_quarter_price_index_fdw.sql new file mode 100644 index 00000000..979f1044 --- /dev/null +++ b/tradein-mvp/backend/data/sql/085_quarter_price_index_fdw.sql @@ -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;