feat(parcels): §22 forecast in analyze endpoint — horizon+enqueue+GET (Step 3b-ii) #1069

Merged
bot-backend merged 1 commit from feat/forecast-endpoint-horizon-enqueue into main 2026-06-05 04:23:28 +00:00
2 changed files with 430 additions and 0 deletions

View file

@ -34,6 +34,7 @@ from app.schemas.parcel import (
from app.services.analysis_runs.repository import (
ANALYZE_SCHEMA_VERSION,
latest_run_dates,
latest_run_for,
persist_analysis_run,
)
from app.services.exporters.layout_tz_pdf import render_layout_tz_pdf
@ -1013,6 +1014,13 @@ def _compute_confidence(
_INLINE_FETCH_WAIT_S = 15
_INLINE_FETCH_POLL_INTERVAL_S = 2
# §22-форсайт (3b-ii): schema_version §22-рана в analysis_runs — это "1.0"
# (SiteFinderReport._SCHEMA_VERSION), отличный от ANALYZE_SCHEMA_VERSION ("analyze-1.0")
# inline-dict analyze. GET /{cad}/forecast читает именно "1.0".
_FORECAST_SCHEMA_VERSION = "1.0"
# Допустимые горизонты прогноза (мес), #995 — иное → 422.
_ALLOWED_FORECAST_HORIZONS = frozenset({6, 12, 18})
@router.get("/by-bbox", response_model=ParcelBboxResponse)
async def get_parcels_by_bbox(
@ -1164,6 +1172,57 @@ def get_fetch_status(
return _fetch_status(db, cad_num)
@router.get("/{cad_num}/forecast")
def get_parcel_forecast(
cad_num: str,
db: Annotated[Session, Depends(get_db)],
response: Response,
x_authenticated_user: Annotated[
str | None,
Header(
alias="X-Authenticated-User",
description="Идентичность из Caddy basic_auth (#994, nullable, read-only здесь)",
),
] = None,
) -> dict[str, Any]:
"""Read-only fetch §22-форсайта участка (3b-ii, #995).
Клиент поллит после POST /analyze (тот fire-and-forget enqueue'ит фон-таску,
считающую §22 SiteFinderReport ~30-180s). Читает ПОСЛЕДНИЙ ран schema_version
"1.0" (§22-форсайт, НЕ analyze-1.0 inline-dict) через schema-filtered
`latest_run_for` (3b-i seam).
Returns:
200 {"status": "ready", "run_id", "created_at", "report"} §22 готов
(report = SiteFinderReport.as_dict() из run.result).
202 {"status": "pending"} ещё не посчитан, клиент продолжает поллить.
Graceful: нет рана 202 pending (НЕ 404/500 на happy "not yet" path);
ошибка БД 202 pending + warning (клиент ретраит, не видит 500).
"""
try:
run = latest_run_for(db, cad_num, schema_version=_FORECAST_SCHEMA_VERSION)
except Exception:
# БД-сбой на read-only поллинге — не валим клиента 500-кой, отдаём pending.
logger.warning(
"forecast read failed for cad=%s — returning pending", cad_num, exc_info=True
)
response.status_code = 202
return {"status": "pending"}
if run is not None:
return {
"status": "ready",
"run_id": run.id,
"created_at": run.created_at,
"report": run.result,
}
# Форсайт ещё не посчитан (таска в работе или /analyze не вызывали) — клиент поллит.
response.status_code = 202
return {"status": "pending"}
@router.post("/{cad_num}/analyze")
def analyze_parcel(
cad_num: str,
@ -1177,6 +1236,10 @@ def analyze_parcel(
str | None,
Query(description="user_id для fallback на default-профиль пользователя"),
] = None,
horizon: Annotated[
int,
Query(description="Горизонт прогноза, мес (#995): один из {6, 12, 18}"),
] = 12,
body: Annotated[
AnalyzeRequest | None,
Body(description="Опциональное тело запроса: inline POI-веса (#201)"),
@ -1205,7 +1268,23 @@ def analyze_parcel(
через GET /fetch-status.
- Дедупликация (через `find_active_on_demand_job`): параллельные запросы
на тот же cad один Celery job, оба клиента ждут.
§22-форсайт (3b-ii, #995): после успешного persist рана best-effort enqueue'им
`forecast_site_finder_report.delay(cad_num, horizon, ...)` fire-and-forget,
провал Celery/Redis НЕ валит ответ (он уже успешен). `horizon` {6, 12, 18}
(иначе 422). Результат добавляется в ответ как `forecast` stub (additive); сам
§22-отчёт клиент забирает через GET /{cad}/forecast (202 pending 200 ready).
"""
# Валидация горизонта прогноза (#995) — до любой работы с БД: иное → 422.
if horizon not in _ALLOWED_FORECAST_HORIZONS:
raise HTTPException(
status_code=422,
detail=(
f"horizon должен быть одним из {sorted(_ALLOWED_FORECAST_HORIZONS)}, "
f"получено: {horizon}"
),
)
# 1) Получить геометрию участка — GeoJSON строка через ST_AsGeoJSON
row = (
db.execute(
@ -2691,6 +2770,28 @@ def analyze_parcel(
created_by=x_authenticated_user,
)
# §22-форсайт (3b-ii, #995): best-effort fire-and-forget enqueue после persist.
# Таска `forecast_site_finder_report` читает только что сохранённый analyze-1.0
# ран и в фоне (~30-180s) считает §22 SiteFinderReport ('1.0'). analyze НЕ ждёт
# её — возвращаемся сразу. Celery/Redis down НЕ должен валить ответ (он уже успешен:
# frontend зависит от формы). Зеркалит best-effort стиль find_or_enqueue_fetch.
# Lazy import — избегаем import-цикла api ↔ workers.tasks на старте.
try:
from app.workers.tasks.forecast import forecast_site_finder_report
forecast_site_finder_report.delay(cad_num, horizon, x_authenticated_user)
result_payload["forecast"] = {"status": "pending", "horizon": horizon}
except Exception:
# Enqueue не удался (broker недоступен и т.п.) — §9.x форсайт advisory,
# клиент узнаёт по status="unavailable" и не будет зря поллить /forecast.
logger.warning(
"forecast enqueue failed for cad=%s horizon=%s — analyze response unaffected",
cad_num,
horizon,
exc_info=True,
)
result_payload["forecast"] = {"status": "unavailable", "horizon": horizon}
return result_payload

View file

@ -0,0 +1,329 @@
"""Тесты §22-форсайта в parcels API (3b-ii, #995).
Покрывает:
`horizon` query-param на POST /analyze: default 12; валидные {6,12,18} приняты;
невалидные (7, 0) 422 ДО любой работы с БД.
Best-effort enqueue после успешного analyze: `forecast_site_finder_report.delay`
вызван ОДИН раз с (cad_num, horizon, created_by); ответ несёт forecast.status="pending".
Enqueue failure: `.delay` бросает analyze всё равно 200, forecast.status="unavailable",
исключение НЕ всплывает.
GET /{cad}/forecast: "1.0"-ран есть (mock latest_run_for) 200 + status=ready + report;
рана нет 202 + status=pending.
Стратегия mock: как test_analyze_market_price.py DB через dependency_overrides,
тяжёлые сервисы патчим. Celery НЕ дёргаем: патчим `.delay` на реальной таске
(lazy import внутри analyze_parcel резолвит атрибут модуля в момент вызова).
RBAC обходится через settings.testing=True (conftest).
"""
from __future__ import annotations
import datetime as dt
from typing import Any
from unittest.mock import MagicMock, patch
from fastapi.testclient import TestClient
from app.main import app
# ── Константы ─────────────────────────────────────────────────────────────────
_CAD = "66:41:0204016:10"
_WKT = "POLYGON((60.6 56.838, 60.61 56.838, 60.61 56.845, 60.6 56.845, 60.6 56.838))"
_GEOJSON = '{"type":"Polygon","coordinates":[[[60.6,56.838],[60.61,56.838]]]}'
# ── Mock factories (зеркалят test_analyze_market_price.py) ──────────────────────
def _make_mapping(data: dict[str, Any]) -> MagicMock:
"""Mock-строка (mapping) с __getitem__ + .get() для dict-like доступа."""
m = MagicMock()
m.__getitem__ = lambda self, k: data[k]
m.get = lambda k, default=None: data.get(k, default)
return m
def _make_db_for_analyze(geom_found: bool = True) -> MagicMock:
"""Mock DB Session, достаточный чтобы analyze_parcel дошёл до persist + enqueue."""
db = MagicMock()
geom_row = (
_make_mapping({"geom_geojson": _GEOJSON, "geom_wkb": None, "source": "cad_quarter"})
if geom_found
else None
)
wkt_row = _make_mapping({"wkt": _WKT}) if geom_found else None
district_row = _make_mapping(
{"district_name": "Октябрьский", "median_price_per_m2": 120000, "dist_to_center": 1500.0}
)
centroid_row = _make_mapping({"lat": 56.84, "lon": 60.605})
def _execute_side_effect(*args: Any, **kwargs: Any) -> MagicMock:
sql = " ".join(str(args[0]).split()) if args else ""
first_val: Any = None
if "AS geom_geojson" in sql:
first_val = geom_row
elif "AS wkt" in sql:
first_val = wkt_row
elif "AS median_price_per_m2" in sql and "district_name" in sql:
first_val = district_row
elif "AS lon" in sql and "AS lat" in sql:
first_val = centroid_row
r = MagicMock()
r.mappings.return_value.first.return_value = first_val
r.mappings.return_value.all.return_value = []
r.scalar.return_value = 0
return r
db.execute.side_effect = _execute_side_effect
ctx = MagicMock()
ctx.__enter__ = MagicMock(return_value=ctx)
ctx.__exit__ = MagicMock(return_value=False)
db.begin_nested.return_value = ctx
return db
def _override_db(db: MagicMock):
def _get_db_override():
yield db
return _get_db_override
# Тяжёлые внешние сервисы — заглушки (как в test_analyze_market_price.py).
_PATCHES = [
patch("app.api.v1.parcels._fetch_air_quality_sync", return_value=None),
patch("app.api.v1.parcels._fetch_weather_sync", return_value=None),
patch("app.api.v1.parcels._fetch_seasonal_weather_sync", return_value=None),
patch(
"app.api.v1.parcels.get_quarter_dump_data",
return_value={
"nspd_zoning": None,
"nspd_zouit_overlaps": [],
"nspd_engineering_nearby": [],
"nspd_dump": {"available": False, "stale": False, "harvest_triggered": False},
},
),
patch("app.api.v1.parcels.compute_velocity", return_value=None),
patch("app.api.v1.parcels.compute_gate_verdict", return_value={"verdict": "unknown"}),
]
def _start_patches() -> None:
for p in _PATCHES:
p.start()
def _stop_patches() -> None:
for p in _PATCHES:
p.stop()
# ── horizon query-param ─────────────────────────────────────────────────────────
def test_horizon_default_12_and_enqueued() -> None:
"""Без horizon → default 12 прокидывается в enqueue, forecast.status=pending."""
from app.core.db import get_db
db = _make_db_for_analyze()
app.dependency_overrides[get_db] = _override_db(db)
_start_patches()
delay_mock = MagicMock()
try:
with patch(
"app.workers.tasks.forecast.forecast_site_finder_report.delay", delay_mock
):
client = TestClient(app)
resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["forecast"] == {"status": "pending", "horizon": 12}
delay_mock.assert_called_once_with(_CAD, 12, None)
finally:
app.dependency_overrides.clear()
_stop_patches()
def test_horizon_valid_values_accepted() -> None:
"""horizon ∈ {6,12,18} принимается → 200 + enqueue с этим горизонтом."""
from app.core.db import get_db
for h in (6, 12, 18):
db = _make_db_for_analyze()
app.dependency_overrides[get_db] = _override_db(db)
_start_patches()
delay_mock = MagicMock()
try:
with patch(
"app.workers.tasks.forecast.forecast_site_finder_report.delay", delay_mock
):
client = TestClient(app)
resp = client.post(f"/api/v1/parcels/{_CAD}/analyze?horizon={h}")
assert resp.status_code == 200, f"horizon={h}: {resp.text}"
assert resp.json()["forecast"] == {"status": "pending", "horizon": h}
delay_mock.assert_called_once_with(_CAD, h, None)
finally:
app.dependency_overrides.clear()
_stop_patches()
def test_horizon_invalid_returns_422() -> None:
"""horizon вне {6,12,18} (7, 0) → 422 ДО enqueue (delay не вызывается)."""
from app.core.db import get_db
for h in (7, 0):
db = _make_db_for_analyze()
app.dependency_overrides[get_db] = _override_db(db)
_start_patches()
delay_mock = MagicMock()
try:
with patch(
"app.workers.tasks.forecast.forecast_site_finder_report.delay", delay_mock
):
client = TestClient(app)
resp = client.post(f"/api/v1/parcels/{_CAD}/analyze?horizon={h}")
assert resp.status_code == 422, f"horizon={h}: {resp.text}"
delay_mock.assert_not_called()
finally:
app.dependency_overrides.clear()
_stop_patches()
def test_enqueue_passes_created_by_header() -> None:
"""X-Authenticated-User прокидывается в .delay как created_by (3-й позиц. арг)."""
from app.core.db import get_db
db = _make_db_for_analyze()
app.dependency_overrides[get_db] = _override_db(db)
_start_patches()
delay_mock = MagicMock()
try:
with patch(
"app.workers.tasks.forecast.forecast_site_finder_report.delay", delay_mock
):
client = TestClient(app)
resp = client.post(
f"/api/v1/parcels/{_CAD}/analyze",
headers={"X-Authenticated-User": "alice"},
)
assert resp.status_code == 200, resp.text
delay_mock.assert_called_once_with(_CAD, 12, "alice")
finally:
app.dependency_overrides.clear()
_stop_patches()
# ── Enqueue failure ──────────────────────────────────────────────────────────
def test_enqueue_failure_returns_200_unavailable() -> None:
""".delay бросает (broker down) → analyze всё равно 200, forecast.status=unavailable."""
from app.core.db import get_db
db = _make_db_for_analyze()
app.dependency_overrides[get_db] = _override_db(db)
_start_patches()
delay_mock = MagicMock(side_effect=RuntimeError("redis down"))
try:
with patch(
"app.workers.tasks.forecast.forecast_site_finder_report.delay", delay_mock
):
client = TestClient(app)
resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["forecast"] == {"status": "unavailable", "horizon": 12}
delay_mock.assert_called_once()
finally:
app.dependency_overrides.clear()
_stop_patches()
def test_analyze_response_keys_unchanged_additive() -> None:
"""forecast — additive: ключевые существующие блоки analyze остаются на месте."""
from app.core.db import get_db
db = _make_db_for_analyze()
app.dependency_overrides[get_db] = _override_db(db)
_start_patches()
try:
with patch("app.workers.tasks.forecast.forecast_site_finder_report.delay", MagicMock()):
client = TestClient(app)
resp = client.post(f"/api/v1/parcels/{_CAD}/analyze")
assert resp.status_code == 200, resp.text
body = resp.json()
# Существующие ключи (до 3b-ii) присутствуют + новый forecast добавлен.
for key in ("cad_num", "district", "score", "score_breakdown", "risks", "forecast"):
assert key in body, f"ключ {key} отсутствует в ответе analyze"
finally:
app.dependency_overrides.clear()
_stop_patches()
# ── GET /{cad}/forecast ──────────────────────────────────────────────────────
def test_get_forecast_ready_when_run_present() -> None:
"""Есть "1.0"-ран → 200 + status=ready + run_id + report=run.result."""
from app.core.db import get_db
report_dict = {"schema_version": "1.0", "deficit": -1.0, "moi": 117}
fake_run = MagicMock()
fake_run.id = 66
fake_run.created_at = dt.datetime(2026, 6, 5, 7, 32, 0)
fake_run.result = report_dict
db = MagicMock()
app.dependency_overrides[get_db] = _override_db(db)
try:
with patch("app.api.v1.parcels.latest_run_for", return_value=fake_run) as lrf:
client = TestClient(app)
resp = client.get(f"/api/v1/parcels/{_CAD}/forecast")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["status"] == "ready"
assert body["run_id"] == 66
assert body["report"] == report_dict
# Читаем именно §22-схему "1.0" (3b-i schema-filtered seam).
_, kwargs = lrf.call_args
assert kwargs.get("schema_version") == "1.0"
finally:
app.dependency_overrides.clear()
def test_get_forecast_pending_when_no_run() -> None:
"""Рана нет (latest_run_for → None) → 202 + status=pending."""
from app.core.db import get_db
db = MagicMock()
app.dependency_overrides[get_db] = _override_db(db)
try:
with patch("app.api.v1.parcels.latest_run_for", return_value=None):
client = TestClient(app)
resp = client.get(f"/api/v1/parcels/{_CAD}/forecast")
assert resp.status_code == 202, resp.text
assert resp.json() == {"status": "pending"}
finally:
app.dependency_overrides.clear()
def test_get_forecast_db_error_returns_pending_not_500() -> None:
"""Ошибка БД на read-only поллинге → 202 pending (graceful, не 500)."""
from app.core.db import get_db
db = MagicMock()
app.dependency_overrides[get_db] = _override_db(db)
try:
with patch(
"app.api.v1.parcels.latest_run_for", side_effect=RuntimeError("db gone")
):
client = TestClient(app)
resp = client.get(f"/api/v1/parcels/{_CAD}/forecast")
assert resp.status_code == 202, resp.text
assert resp.json() == {"status": "pending"}
finally:
app.dependency_overrides.clear()