feat(tradein): Phase 3.2 — /api/v1/search endpoint + Redis cache

Multi-Source Integration Phase 3.2. /api/v1/search with cross-source filters
+ POST body params + Redis 5min hot cache.

New files:
  app/schemas/search.py            — SearchParams (~20 filters, Pydantic v2)
  app/schemas/search_response.py   — SearchResponse + SearchResultItem
  app/services/search_query.py     — build_search_query (psycopg v3, CAST(:x AS type))
  app/services/cache.py            — SearchCache async wrapper (singleton lru_cache)
  app/api/v1/search.py             — POST /search endpoint
  tests/test_search_api.py         — unit + integration tests

Edited:
  app/main.py                      — include search.router prefix=/api/v1
  app/core/config.py               — settings.redis_url default
  pyproject.toml                   — add redis>=5.0.0

Queries listings_search_mv (matview from PR #469). Cache failures swallowed
(degrade to slow-but-correct). Sort whitelisted dict (no SQL injection).

Refs Master Plan sec 9.1 + 7.2.

tradein-mvp/backend/app/api/v1/search.py

@@ -0,0 +1,64 @@
+"""Search endpoint — POST /api/v1/search (Phase 3.2, master plan sec 9.1)."""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import Annotated
+
+from fastapi import APIRouter, Depends
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.db import get_db
+from app.schemas.search import SearchParams
+from app.schemas.search_response import SearchResponse, SearchResultItem
+from app.services.cache import get_search_cache
+from app.services.search_query import build_count_query, build_search_query
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+@router.post("/search", response_model=SearchResponse)
+async def search(
+    params: SearchParams,
+    db: Annotated[Session, Depends(get_db)],
+) -> SearchResponse:
+    """Поиск listings по фильтрам (cross-source matview + Redis 5min cache)."""
+    started = time.monotonic()
+    cache = get_search_cache()
+    cache_key = cache.search_key(params.model_dump(mode="json"))
+
+    cached = await cache.get(cache_key)
+    if cached is not None:
+        logger.info(
+            "search cache HIT key=%s page=%d size=%d",
+            cache_key[:24], params.page, params.page_size,
+        )
+        return SearchResponse.model_validate(cached)
+
+    sql, args = build_search_query(params)
+    rows = db.execute(text(sql), args).mappings().all()
+
+    count_sql, count_args = build_count_query(params)
+    total = int(db.execute(text(count_sql), count_args).scalar() or 0)
+
+    items = [SearchResultItem.model_validate(dict(r)) for r in rows]
+    elapsed_ms = (time.monotonic() - started) * 1000.0
+    response = SearchResponse(
+        items=items,
+        total=total,
+        page=params.page,
+        page_size=params.page_size,
+        elapsed_ms=round(elapsed_ms, 1),
+        cache_hit=False,
+    )
+
+    await cache.set(cache_key, response.model_dump(mode="json"), ttl=cache.TTL_SEARCH)
+    logger.info(
+        "search MISS page=%d size=%d total=%d items=%d elapsed=%.1fms",
+        params.page, params.page_size, total, len(items), elapsed_ms,
+    )
+    return response

tradein-mvp/backend/app/core/config.py

@@ -29,5 +29,8 @@ class Settings(BaseSettings):
     # Задаётся через env COOKIE_ENCRYPTION_KEY. Пусто = шифрование не работает.
     cookie_encryption_key: str = ""
 
+    # Redis URL для hot-cache (Phase 3.2). Задаётся через env REDIS_URL.
+    redis_url: str = "redis://localhost:6379/0"
+
 
 settings = Settings()

tradein-mvp/backend/app/main.py

@@ -12,7 +12,7 @@ import sentry_sdk
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
-from app.api.v1 import admin, brand, geocode, trade_in
+from app.api.v1 import admin, brand, geocode, search, trade_in
 from app.core.config import settings
 from app.core.ratelimit import RateLimitMiddleware
 
@@ -58,3 +58,4 @@ app.include_router(geocode.router, prefix="/api/v1/geocode", tags=["geocode"])
 app.include_router(admin.router, prefix="/api/v1/admin", tags=["admin"])
 app.include_router(brand.router, prefix="/api/v1/brand", tags=["brand"])
 app.include_router(trade_in.router, prefix="/api/v1/trade-in", tags=["trade-in"])
+app.include_router(search.router, prefix="/api/v1", tags=["search"])

tradein-mvp/backend/app/schemas/search.py

@@ -0,0 +1,71 @@
+"""SearchParams + SearchResponse — /api/v1/search (Phase 3.2)."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+SortKey = Literal[
+    "price_asc", "price_desc", "area_desc", "area_asc", "date_desc", "dist_asc"
+]
+
+
+class SearchParams(BaseModel):
+    """Фильтры поиска по listings_search_mv (master plan sec 9.1)."""
+
+    # --- Geo (radius search) ---
+    lat: float | None = Field(default=None, ge=-90.0, le=90.0)
+    lon: float | None = Field(default=None, ge=-180.0, le=180.0)
+    radius_m: int = Field(default=2000, ge=100, le=50000)
+
+    # --- Property filters ---
+    rooms: int | None = Field(default=None, ge=0, le=10)
+    rooms_in: list[int] | None = None
+    area_m2_min: float | None = Field(default=None, ge=0)
+    area_m2_max: float | None = Field(default=None, ge=0)
+    price_rub_min: int | None = Field(default=None, ge=0)
+    price_rub_max: int | None = Field(default=None, ge=0)
+    price_per_m2_max: int | None = Field(default=None, ge=0)
+    floor_min: int | None = Field(default=None, ge=1)
+    floor_max: int | None = Field(default=None, ge=1)
+
+    # --- House filters ---
+    year_built_min: int | None = Field(default=None, ge=1800, le=2100)
+    year_built_max: int | None = Field(default=None, ge=1800, le=2100)
+    house_class: list[str] | None = None
+    floors_total_min: int | None = Field(default=None, ge=1)
+    floors_total_max: int | None = Field(default=None, ge=1)
+
+    # --- Quality / cross-source ---
+    has_kadastr: bool = False
+    sources: list[Literal["avito", "cian", "yandex_realty", "n1"]] | None = None
+    multi_source_only: bool = False
+    require_avito: bool = False
+    require_cian: bool = False
+    require_yandex: bool = False
+
+    # --- Text search ---
+    address_query: str | None = Field(default=None, max_length=200)
+    description_query: str | None = Field(default=None, max_length=200)
+
+    # --- Sort + pagination ---
+    sort: SortKey = "date_desc"
+    page: int = Field(default=1, ge=1, le=1000)
+    page_size: int = Field(default=50, ge=1, le=200)
+
+    @model_validator(mode="after")
+    def _validate_geo_pair(self) -> SearchParams:
+        if (self.lat is None) != (self.lon is None):
+            raise ValueError("lat and lon must be provided together")
+        if self.sort == "dist_asc" and self.lat is None:
+            raise ValueError("sort=dist_asc requires lat+lon")
+        if self.price_rub_min and self.price_rub_max and self.price_rub_min > self.price_rub_max:
+            raise ValueError("price_rub_min > price_rub_max")
+        if self.area_m2_min and self.area_m2_max and self.area_m2_min > self.area_m2_max:
+            raise ValueError("area_m2_min > area_m2_max")
+        return self
+
+    @property
+    def offset(self) -> int:
+        return (self.page - 1) * self.page_size

tradein-mvp/backend/app/schemas/search_response.py

@@ -0,0 +1,53 @@
+"""SearchResponse + SearchResultItem — /api/v1/search return shape."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+
+class SearchResultItem(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    listing_id: int
+    source: str
+    source_url: str | None = None
+    address: str | None = None
+    lat: float | None = None
+    lng: float | None = None
+
+    rooms: int | None = None
+    total_area: float | None = None
+    floor: int | None = None
+    total_floors: int | None = None
+
+    price_rub: int | None = None
+    price_per_m2: int | None = None
+    kadastr_num: str | None = None
+    scraped_at: datetime | None = None
+
+    house_id: int | None = None
+    year_built: int | None = None
+    house_class: str | None = None
+    developer_name: str | None = None
+    house_rating: float | None = None
+    house_ratings_count: int | None = None
+
+    source_count: int | None = None
+    sources: list[str] | None = None
+    has_avito: bool | None = None
+    has_cian: bool | None = None
+    has_yandex: bool | None = None
+
+    house_median_ppm2: float | None = None
+    district: str | None = None
+
+
+class SearchResponse(BaseModel):
+    items: list[SearchResultItem]
+    total: int
+    page: int
+    page_size: int
+    elapsed_ms: float | None = None
+    cache_hit: bool = False

tradein-mvp/backend/app/services/cache.py

@@ -0,0 +1,64 @@
+"""Redis hot cache for search results (master plan sec 7.2)."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+from functools import lru_cache
+from typing import Any
+
+import redis.asyncio as redis_async
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+
+class SearchCache:
+    """Async Redis cache wrapper. Stable sha256 hash keys, TTL per kind."""
+
+    TTL_SEARCH: int = 300
+    TTL_HOUSE_DETAIL: int = 3600
+    TTL_VALUATION: int = 86400
+
+    def __init__(self, url: str) -> None:
+        self._client: redis_async.Redis = redis_async.from_url(
+            url, decode_responses=True, socket_timeout=2.0, socket_connect_timeout=2.0
+        )
+
+    def search_key(self, params: dict[str, Any]) -> str:
+        payload = json.dumps(params, sort_keys=True, default=str, ensure_ascii=False)
+        digest = hashlib.sha256(payload.encode("utf-8")).hexdigest()[:32]
+        return f"tradein:search:{digest}"
+
+    async def get(self, key: str) -> dict[str, Any] | None:
+        try:
+            raw = await self._client.get(key)
+        except Exception as e:
+            logger.warning("redis GET failed key=%s: %s", key[:24], e)
+            return None
+        if raw is None:
+            return None
+        try:
+            return json.loads(raw)
+        except json.JSONDecodeError:
+            logger.warning("redis cache corrupted key=%s", key[:24])
+            return None
+
+    async def set(self, key: str, value: dict[str, Any], ttl: int) -> None:
+        try:
+            await self._client.set(
+                key, json.dumps(value, default=str, ensure_ascii=False), ex=ttl
+            )
+        except Exception as e:
+            logger.warning("redis SET failed key=%s: %s", key[:24], e)
+
+    async def close(self) -> None:
+        await self._client.aclose()
+
+
+@lru_cache(maxsize=1)
+def get_search_cache() -> SearchCache:
+    """Singleton — один Redis pool на процесс."""
+    return SearchCache(settings.redis_url)

tradein-mvp/backend/app/services/search_query.py

@@ -0,0 +1,138 @@
+"""SQL builder for /api/v1/search — parameterized, psycopg v3 (CAST(:x AS type))."""
+
+from __future__ import annotations
+
+from app.schemas.search import SearchParams
+
+_SORT_SQL: dict[str, str] = {
+    "price_asc": "price_rub ASC NULLS LAST",
+    "price_desc": "price_rub DESC NULLS LAST",
+    "area_desc": "total_area DESC NULLS LAST",
+    "area_asc": "total_area ASC NULLS LAST",
+    "date_desc": "scraped_at DESC NULLS LAST",
+    "dist_asc": (
+        "ST_Distance(geom::geography, "
+        "ST_MakePoint(CAST(:lon AS double precision), "
+        "CAST(:lat AS double precision))::geography) ASC"
+    ),
+}
+
+
+def build_search_query(params: SearchParams) -> tuple[str, dict[str, object]]:
+    """Возвращает (sql, args) для SELECT из listings_search_mv."""
+    where: list[str] = ["1=1"]
+    args: dict[str, object] = {}
+
+    if params.lat is not None and params.lon is not None:
+        where.append(
+            "ST_DWithin(geom::geography, "
+            "ST_MakePoint(CAST(:lon AS double precision), "
+            "CAST(:lat AS double precision))::geography, "
+            "CAST(:radius_m AS integer))"
+        )
+        args["lat"] = params.lat
+        args["lon"] = params.lon
+        args["radius_m"] = params.radius_m
+
+    if params.rooms is not None:
+        where.append("rooms = CAST(:rooms AS integer)")
+        args["rooms"] = params.rooms
+    if params.rooms_in:
+        where.append("rooms = ANY(CAST(:rooms_in AS integer[]))")
+        args["rooms_in"] = params.rooms_in
+
+    if params.area_m2_min is not None:
+        where.append("total_area >= CAST(:area_min AS double precision)")
+        args["area_min"] = params.area_m2_min
+    if params.area_m2_max is not None:
+        where.append("total_area <= CAST(:area_max AS double precision)")
+        args["area_max"] = params.area_m2_max
+
+    if params.price_rub_min is not None:
+        where.append("price_rub >= CAST(:price_min AS bigint)")
+        args["price_min"] = params.price_rub_min
+    if params.price_rub_max is not None:
+        where.append("price_rub <= CAST(:price_max AS bigint)")
+        args["price_max"] = params.price_rub_max
+    if params.price_per_m2_max is not None:
+        where.append("price_per_m2 <= CAST(:ppm2_max AS bigint)")
+        args["ppm2_max"] = params.price_per_m2_max
+
+    if params.floor_min is not None:
+        where.append("floor >= CAST(:floor_min AS integer)")
+        args["floor_min"] = params.floor_min
+    if params.floor_max is not None:
+        where.append("floor <= CAST(:floor_max AS integer)")
+        args["floor_max"] = params.floor_max
+
+    if params.year_built_min is not None:
+        where.append("year_built >= CAST(:yb_min AS integer)")
+        args["yb_min"] = params.year_built_min
+    if params.year_built_max is not None:
+        where.append("year_built <= CAST(:yb_max AS integer)")
+        args["yb_max"] = params.year_built_max
+
+    if params.house_class:
+        where.append("house_class = ANY(CAST(:hclass AS text[]))")
+        args["hclass"] = params.house_class
+
+    if params.floors_total_min is not None:
+        where.append("total_floors >= CAST(:fl_total_min AS integer)")
+        args["fl_total_min"] = params.floors_total_min
+    if params.floors_total_max is not None:
+        where.append("total_floors <= CAST(:fl_total_max AS integer)")
+        args["fl_total_max"] = params.floors_total_max
+
+    if params.has_kadastr:
+        where.append("kadastr_num IS NOT NULL")
+
+    if params.sources:
+        where.append("sources && CAST(:sources AS text[])")
+        args["sources"] = params.sources
+    if params.multi_source_only:
+        where.append("source_count >= 2")
+    if params.require_avito:
+        where.append("has_avito = true")
+    if params.require_cian:
+        where.append("has_cian = true")
+    if params.require_yandex:
+        where.append("has_yandex = true")
+
+    if params.address_query:
+        where.append("address ILIKE CAST(:addr_like AS text)")
+        args["addr_like"] = f"%{params.address_query}%"
+    if params.description_query:
+        where.append("tsv @@ plainto_tsquery('russian', CAST(:descq AS text))")
+        args["descq"] = params.description_query
+
+    where_sql = " AND ".join(where)
+    order_sql = _SORT_SQL[params.sort]
+
+    sql = (
+        "SELECT listing_id, source, source_url, address, lat, lng, "
+        "rooms, total_area, floor, total_floors, price_rub, price_per_m2, "
+        "kadastr_num, scraped_at, "
+        "house_id, year_built, house_class, developer_name, "
+        "house_rating, house_ratings_count, "
+        "source_count, sources, has_avito, has_cian, has_yandex, "
+        "house_median_ppm2, district "
+        "FROM listings_search_mv "
+        f"WHERE {where_sql} "
+        f"ORDER BY {order_sql} "
+        "LIMIT CAST(:limit AS integer) OFFSET CAST(:offset AS integer)"
+    )
+    args["limit"] = params.page_size
+    args["offset"] = params.offset
+
+    return sql, args
+
+
+def build_count_query(params: SearchParams) -> tuple[str, dict[str, object]]:
+    """COUNT(*) — отдельный запрос для total."""
+    sql, args = build_search_query(params)
+    where_start = sql.find("WHERE ")
+    order_start = sql.find(" ORDER BY ")
+    where_clause = sql[where_start:order_start]
+    count_args = {k: v for k, v in args.items() if k not in ("limit", "offset")}
+    count_sql = f"SELECT count(*) AS total FROM listings_search_mv {where_clause}"
+    return count_sql, count_args

tradein-mvp/backend/pyproject.toml

@@ -20,6 +20,7 @@ dependencies = [
     "curl-cffi>=0.7.0",           # impersonate=chrome120 для Cian/Avito (TLS fingerprint)
     "python-multipart>=0.0.9",    # FastAPI UploadFile — загрузка фото квартиры (#394)
     "sentry-sdk>=2.0.0",          # мониторинг ошибок → GlitchTip (#396)
+    "redis>=5.0.0",               # async hot cache для /api/v1/search (Phase 3.2)
 ]
 
 [dependency-groups]

tradein-mvp/backend/tests/test_search_api.py

@@ -0,0 +1,110 @@
+"""Tests for /api/v1/search — golden path + filters + cache hit/miss."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.main import app
+from app.schemas.search import SearchParams
+from app.services.cache import get_search_cache
+from app.services.search_query import build_count_query, build_search_query
+
+
+def test_build_query_minimal():
+    sql, args = build_search_query(SearchParams())
+    assert "FROM listings_search_mv" in sql
+    assert "WHERE 1=1" in sql
+    assert "scraped_at DESC" in sql
+    assert args["limit"] == 50
+    assert args["offset"] == 0
+
+
+def test_build_query_geo_radius():
+    sql, args = build_search_query(
+        SearchParams(lat=56.838, lon=60.594, radius_m=1500, sort="dist_asc")
+    )
+    assert "ST_DWithin" in sql
+    assert "CAST(:lat AS double precision)" in sql
+    assert args["lat"] == 56.838
+    assert args["radius_m"] == 1500
+
+
+def test_build_query_all_filters():
+    params = SearchParams(
+        rooms=2, area_m2_min=40, area_m2_max=80, price_rub_max=10_000_000,
+        year_built_min=2000, has_kadastr=True, multi_source_only=True,
+        require_avito=True, address_query="Малышева",
+    )
+    sql, args = build_search_query(params)
+    assert "rooms = CAST(:rooms AS integer)" in sql
+    assert "total_area >= CAST(:area_min AS double precision)" in sql
+    assert "price_rub <= CAST(:price_max AS bigint)" in sql
+    assert "year_built >= CAST(:yb_min AS integer)" in sql
+    assert "kadastr_num IS NOT NULL" in sql
+    assert "source_count >= 2" in sql
+    assert "has_avito = true" in sql
+    assert "address ILIKE" in sql
+    assert args["addr_like"] == "%Малышева%"
+
+
+def test_build_query_sources_array():
+    sql, args = build_search_query(SearchParams(sources=["avito", "cian"]))
+    assert "sources && CAST(:sources AS text[])" in sql
+    assert args["sources"] == ["avito", "cian"]
+
+
+def test_build_count_query_strips_limit():
+    sql, args = build_count_query(SearchParams(rooms=2, page=3))
+    assert "count(*)" in sql
+    assert "ORDER BY" not in sql
+    assert "LIMIT" not in sql
+    assert "offset" not in args
+    assert "limit" not in args
+
+
+def test_geo_pair_required():
+    with pytest.raises(ValueError, match="lat and lon"):
+        SearchParams(lat=56.0)
+
+
+def test_dist_sort_requires_geo():
+    with pytest.raises(ValueError, match="dist_asc"):
+        SearchParams(sort="dist_asc")
+
+
+def test_price_range_consistency():
+    with pytest.raises(ValueError, match="price_rub_min > price_rub_max"):
+        SearchParams(price_rub_min=10_000_000, price_rub_max=5_000_000)
+
+
+def test_page_size_capped():
+    with pytest.raises(ValueError):
+        SearchParams(page_size=500)
+
+
+@pytest.fixture(autouse=True)
+def _reset_cache_singleton(monkeypatch):
+    get_search_cache.cache_clear()
+    mock = AsyncMock()
+    mock.get = AsyncMock(return_value=None)
+    mock.set = AsyncMock()
+    mock.search_key = lambda p: "tradein:search:test"
+    mock.TTL_SEARCH = 300
+    monkeypatch.setattr("app.api.v1.search.get_search_cache", lambda: mock)
+    return mock
+
+
+def test_search_cache_hit(_reset_cache_singleton):
+    cached_payload = {
+        "items": [], "total": 0, "page": 1, "page_size": 50,
+        "elapsed_ms": 0.0, "cache_hit": True,
+    }
+    _reset_cache_singleton.get = AsyncMock(return_value=cached_payload)
+    client = TestClient(app)
+    r = client.post("/api/v1/search", json={"rooms": 2})
+    assert r.status_code == 200
+    assert r.json()["cache_hit"] is True
+    _reset_cache_singleton.set.assert_not_awaited()