lekss361
e46b7d778e
Multi-Source Integration Phase 3.1. Foundation for /api/v1/search p95 < 100ms on 100K+ listings. SQL 050_search_optimization.sql: - CREATE EXTENSION IF NOT EXISTS pg_trgm - ALTER listings ADD COLUMN tsv tsvector GENERATED ALWAYS AS (...) STORED - 10 indexes (geom GIST, composite, gin_trgm, GIN tsv, partial kadastr, ...) - CREATE MATERIALIZED VIEW listings_search_mv (per-listing aggregated across sources) - 5 matview indexes (UNIQUE on listing_id for REFRESH CONCURRENTLY) Celery task tradein.refresh_search_matview: - REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv (psycopg v3) - Logged duration via logger.info - Graceful try/except ImportError fallback (no celery_app.py in tradein-mvp yet) - Beat schedule snippet in docstring for future wiring Fixups (deep-code-reviewer 2026-05-23): - h.ratings_count → h.reviews_count (alias house_ratings_count preserved) - Strip +psycopg dialect prefix from DSN for libpq compat Refs Master Plan sec 5.1 + 6.5 + 10.1.
70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
"""Refresh listings_search_mv materialized view (daily 3 AM).
|
|
|
|
Currently tradein-mvp has no Celery infrastructure (verified — no celery_app.py exists
|
|
in tradein-mvp/backend/app/). This module provides both:
|
|
- A Celery task wrapper (active only if celery is installed and celery_app present).
|
|
- A plain async refresh_search_matview() callable for cron / one-shot invocation.
|
|
|
|
NEEDS COORDINATION: main session must decide whether to bootstrap Celery in tradein-mvp
|
|
or run via external scheduler (systemd timer / OS cron). Until then, no Beat schedule.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
|
|
import psycopg
|
|
|
|
from app.core.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def refresh_search_matview() -> None:
|
|
"""REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv (psycopg v3, sync).
|
|
|
|
Logs duration. Idempotent. Safe to run during read traffic (CONCURRENTLY).
|
|
"""
|
|
start = time.monotonic()
|
|
# DATABASE_URL is SQLAlchemy dialect form (postgresql+psycopg://...) in tradein-mvp
|
|
# (see tradein-mvp/docker-compose.prod.yml). libpq / psycopg.connect() accepts only
|
|
# postgresql:// or postgres:// — strip the +psycopg dialect prefix.
|
|
dsn = settings.database_url.replace("postgresql+psycopg://", "postgresql://", 1)
|
|
with psycopg.connect(dsn, autocommit=True) as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute("REFRESH MATERIALIZED VIEW CONCURRENTLY listings_search_mv")
|
|
elapsed = time.monotonic() - start
|
|
logger.info("listings_search_mv refresh completed in %.2fs", elapsed)
|
|
|
|
|
|
# Celery-task wrapper — only registers if celery_app exists.
|
|
try:
|
|
from app.celery_app import celery_app # type: ignore[import-not-found]
|
|
except ImportError:
|
|
logger.info(
|
|
"Celery not configured in tradein-mvp — refresh_search_matview() callable directly. "
|
|
"Bootstrap app/celery_app.py to enable scheduled execution.",
|
|
)
|
|
else:
|
|
|
|
@celery_app.task(name="tradein.refresh_search_matview") # type: ignore[misc]
|
|
def refresh_search_matview_task() -> None:
|
|
"""Celery wrapper for refresh_search_matview()."""
|
|
refresh_search_matview()
|
|
|
|
|
|
# =============================================================================
|
|
# Beat schedule snippet — paste into app/celery_app.py if/when Celery is bootstrapped:
|
|
#
|
|
# from celery.schedules import crontab
|
|
#
|
|
# celery_app.conf.beat_schedule = {
|
|
# 'tradein-refresh-matview': {
|
|
# 'task': 'tradein.refresh_search_matview',
|
|
# 'schedule': crontab(hour=3, minute=0),
|
|
# },
|
|
# # ... other tradein beat entries
|
|
# }
|
|
#
|
|
# Per master plan sec 10.1 ('refresh-search-matview-daily': crontab(minute=0, hour=3)).
|
|
# =============================================================================
|