"""Refresh helper for the sales-tracker MVs (Issue #61). Two independent materialized views built from the Объектив sales-tracker ("шахматки") snapshots (objective_lots / objective_lots_history), created by data/sql/164_mv_sales_tracker_velocity_absorption.sql: 1. mv_sales_tracker_velocity_by_district — per (district, month) sold/total/ avg-sold-price. Feeds the Site Finder Velocity Score (4th scoring criterion). 2. mv_sales_tracker_absorption_curves — cumulative sold% as f(months from sales_start_date) per (rooms_int, area_bucket). Foundation for recommend_mix + sellout forecast. The two MVs do not depend on each other, so refresh order is irrelevant; both are refreshed in the same call. Scheduled via Celery beat hardcoded entry in workers/beat_schedule.py ('mv-sales-tracker-refresh-weekly', Mon 04:30 MSK). Usage example (manual, via psql-connected shell or admin endpoint): from app.services.site_finder.sales_tracker_mv_refresh import refresh_sales_tracker_mvs counts = refresh_sales_tracker_mvs(db) # logs: "mv_sales_tracker_velocity_by_district refreshed: 70 rows", etc. """ from __future__ import annotations import logging from sqlalchemy import text from sqlalchemy.exc import DatabaseError from sqlalchemy.orm import Session logger = logging.getLogger(__name__) _MV_NAMES: tuple[str, ...] = ( "mv_sales_tracker_velocity_by_district", "mv_sales_tracker_absorption_curves", ) def _refresh_mv(db: Session, mv_name: str, *, concurrently: bool) -> int: """Run REFRESH MATERIALIZED VIEW [CONCURRENTLY] , return row count. Falls back to non-concurrent on the known "cannot refresh concurrently" error (MV empty or no UNIQUE index — should not happen in prod since the migration creates the UNIQUE index and populates the MV, but provides a safe recovery path for first-run / post-recreation 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 DatabaseError as e: # PostgreSQL emits "CONCURRENTLY cannot be used when the materialized # view ... is not populated" (matview.c, SQLSTATE 55000), surfaced by # psycopg3 as an InternalError (a DatabaseError sibling). if concurrently and "concurrently cannot be used" 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 row = db.execute(text(f"SELECT COUNT(*) FROM {mv_name}")).first() count = int(row[0]) if row else 0 logger.info("%s refreshed: %d rows", mv_name, count) return count def refresh_sales_tracker_mvs(db: Session, *, concurrently: bool = True) -> dict[str, int]: """Refresh both sales-tracker MVs. Args: db: SQLAlchemy Session (sync). concurrently: When True, uses REFRESH CONCURRENTLY (non-blocking — readers continue). Requires the per-MV UNIQUE indexes (mv_sales_tracker_velocity_by_district_pk, mv_sales_tracker_absorption_curves_pk) and the MVs to be already populated. Pass False only for first populate or after recreation. Returns: Mapping mv_name -> row count after refresh (for observability). """ counts: dict[str, int] = {} for mv_name in _MV_NAMES: counts[mv_name] = _refresh_mv(db, mv_name, concurrently=concurrently) return counts