风控写接口接入平台幂等（docs/25 P3 #22）
docs/05 §5.1 把"业务写接口"列为必须携带 Idempotency-Key 的接口，风控 6 个 POST （手工扫描、确认接收、进入调查、关闭误报、完成结案、升级处理）此前一个都没带， 重复提交会二次驱动状态机。 复用平台的 api_request_receipt 与 ApiTransactionService，但新增 execute_in：原来的 execute 自己开 SessionFactory() 和 session.begin()，而 RiskActionService._finish 会在内部 commit，套进去就成了"内层提交外层事务"。execute_in 改为在调用方传入的 session 上读写幂等记录，幂等记录因此与业务写入同处一个事务（§5.2）。 scope 用实际路径（含 alert_no）而不是路由模板：§5.1 的幂等范围是 user_id + method + normalized_path + idempotency_key，把路径参数折成模板会让同一个键 在不同预警之间互相回放 —— 那是把两次不同资源的操作当成一次。 测试：单测加幂等透传替身与"缺键即 422"用例；新增 tests/integration/test_risk_idempotency_mysql.py，覆盖同键回放不重复执行、同键不同 正文 409、缺键/非 ASCII 拒绝，以及端到端"重复 POST 只调用一次处置逻辑"。
This commit is contained in:
+82
-15
@@ -1,11 +1,11 @@
|
||||
"""风控只读查询接口。"""
|
||||
|
||||
import json
|
||||
from collections.abc import AsyncIterator
|
||||
from collections.abc import AsyncIterator, Awaitable, Callable
|
||||
from datetime import datetime, time
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Depends, File, Path, Request, UploadFile
|
||||
from fastapi import APIRouter, Depends, File, Header, Path, Request, UploadFile
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from starlette.responses import StreamingResponse
|
||||
|
||||
@@ -27,6 +27,7 @@ from app.api.schemas.risk import (
|
||||
from app.core.contracts import RequestContext
|
||||
from app.core.errors import SseNotAcceptableError
|
||||
from app.infrastructure.db import mysql_scan_lock
|
||||
from app.service.api_transaction_service import ApiTransactionService
|
||||
from app.service.risk_action_service import RiskActionService
|
||||
from app.service.risk_daily_report_mail_service import RiskDailyReportMailService
|
||||
from app.service.risk_daily_report_service import RiskDailyReportService
|
||||
@@ -42,6 +43,26 @@ router = APIRouter(
|
||||
)
|
||||
|
||||
|
||||
async def _idempotent_write(
|
||||
session: AsyncSession,
|
||||
context: RequestContext,
|
||||
key: str | None,
|
||||
scope: str,
|
||||
body: Any,
|
||||
action: Callable[[AsyncSession], Awaitable[dict[str, Any]]],
|
||||
) -> dict[str, Any]:
|
||||
"""风控写接口的统一幂等入口（`docs/05` §5.1、§5.2、§13.2）。
|
||||
|
||||
`scope` 用**实际**路径（含 `alert_no`）：§5.1 的幂等范围是
|
||||
`user_id + method + normalized_path + idempotency_key`，把路径参数折成模板会让
|
||||
同一个键在不同预警之间互相回放 —— 那是把两次不同资源的操作当成一次。
|
||||
|
||||
幂等记录与业务写入同事务（见 `ApiTransactionService.execute_in`），重复请求直接
|
||||
回放 `response_json`，不会二次驱动状态机。
|
||||
"""
|
||||
return await ApiTransactionService().execute_in(session, context, scope, key, body, action)
|
||||
|
||||
|
||||
@router.get("/overview")
|
||||
async def risk_overview(
|
||||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||||
@@ -65,15 +86,21 @@ async def list_risk_alerts(
|
||||
async def scan_risk_alerts(
|
||||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||||
) -> dict[str, object]:
|
||||
# 手工触发的扫描必须与定时扫描互斥，否则两条路径会同时查不到重复、同时插入。
|
||||
# 锁加在**入口层**而不是 `RiskScanService.scan()` 内部：`GET_LOCK` 是连接级的，
|
||||
# 而调度器已在它自己的 session 上持锁 —— 被两个入口共用的服务方法若再取同一把锁，
|
||||
# 取锁的连接不是持锁的那一个、必然失败，会**把定时扫描自己挡死**。
|
||||
async with mysql_scan_lock() as acquired:
|
||||
if not acquired:
|
||||
raise RiskScanBusyError("规则扫描正在执行，请稍后重试")
|
||||
data = await RiskScanService(session).scan(context)
|
||||
async def run(inner: AsyncSession) -> dict[str, Any]:
|
||||
# 手工触发的扫描必须与定时扫描互斥，否则两条路径会同时查不到重复、同时插入。
|
||||
# 锁加在**入口层**而不是 `RiskScanService.scan()` 内部：`GET_LOCK` 是连接级的，
|
||||
# 而调度器已在它自己的 session 上持锁 —— 被两个入口共用的服务方法若再取同一把锁，
|
||||
# 取锁的连接不是持锁的那一个、必然失败，会**把定时扫描自己挡死**。
|
||||
async with mysql_scan_lock() as acquired:
|
||||
if not acquired:
|
||||
raise RiskScanBusyError("规则扫描正在执行，请稍后重试")
|
||||
return await RiskScanService(inner).scan(context)
|
||||
|
||||
data = await _idempotent_write(
|
||||
session, context, key, "POST /api/v1/risk/alerts/scan", {}, run
|
||||
)
|
||||
return _envelope(data, context)
|
||||
|
||||
|
||||
@@ -82,8 +109,16 @@ async def acknowledge_risk_alert(
|
||||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||||
) -> dict[str, object]:
|
||||
data = await RiskActionService(session).acknowledge(alert_no, context)
|
||||
data = await _idempotent_write(
|
||||
session,
|
||||
context,
|
||||
key,
|
||||
f"POST /api/v1/risk/alerts/{alert_no}/acknowledgements",
|
||||
{},
|
||||
lambda inner: RiskActionService(inner).acknowledge(alert_no, context),
|
||||
)
|
||||
return _envelope(data, context)
|
||||
|
||||
|
||||
@@ -92,8 +127,16 @@ async def investigate_risk_alert(
|
||||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||||
) -> dict[str, object]:
|
||||
data = await RiskActionService(session).investigate(alert_no, context)
|
||||
data = await _idempotent_write(
|
||||
session,
|
||||
context,
|
||||
key,
|
||||
f"POST /api/v1/risk/alerts/{alert_no}/investigations",
|
||||
{},
|
||||
lambda inner: RiskActionService(inner).investigate(alert_no, context),
|
||||
)
|
||||
return _envelope(data, context)
|
||||
|
||||
|
||||
@@ -103,8 +146,16 @@ async def exclude_risk_alert(
|
||||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||||
) -> dict[str, object]:
|
||||
data = await RiskActionService(session).exclude(alert_no, payload.reason, context)
|
||||
data = await _idempotent_write(
|
||||
session,
|
||||
context,
|
||||
key,
|
||||
f"POST /api/v1/risk/alerts/{alert_no}/exclusions",
|
||||
{"reason": payload.reason},
|
||||
lambda inner: RiskActionService(inner).exclude(alert_no, payload.reason, context),
|
||||
)
|
||||
return _envelope(data, context)
|
||||
|
||||
|
||||
@@ -114,8 +165,16 @@ async def resolve_risk_alert(
|
||||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||||
) -> dict[str, object]:
|
||||
data = await RiskActionService(session).resolve(alert_no, payload.resolution, context)
|
||||
data = await _idempotent_write(
|
||||
session,
|
||||
context,
|
||||
key,
|
||||
f"POST /api/v1/risk/alerts/{alert_no}/resolutions",
|
||||
{"resolution": payload.resolution},
|
||||
lambda inner: RiskActionService(inner).resolve(alert_no, payload.resolution, context),
|
||||
)
|
||||
return _envelope(data, context)
|
||||
|
||||
|
||||
@@ -125,8 +184,16 @@ async def escalate_risk_alert(
|
||||
alert_no: str = Path(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$"),
|
||||
context: RequestContext = Depends(build_request_context), # noqa: B008
|
||||
session: AsyncSession = Depends(get_session), # noqa: B008
|
||||
key: str | None = Header(default=None, alias="Idempotency-Key"),
|
||||
) -> dict[str, object]:
|
||||
data = await RiskActionService(session).escalate(alert_no, payload.reason, context)
|
||||
data = await _idempotent_write(
|
||||
session,
|
||||
context,
|
||||
key,
|
||||
f"POST /api/v1/risk/alerts/{alert_no}/escalations",
|
||||
{"reason": payload.reason},
|
||||
lambda inner: RiskActionService(inner).escalate(alert_no, payload.reason, context),
|
||||
)
|
||||
return _envelope(data, context)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user