181 lines
7.7 KiB
Python
181 lines
7.7 KiB
Python
"""routers/bioflow_proxy.py —— BFF 生信反代（FR-BIO-027/028：BFF 仅流式反代 + 报告登记）。
|
||
|
||
- 流式反代 /api/bioflow/* → BIOFLOW_SERVICE_URL（SSE 不缓冲透传：
|
||
httpx client.stream 逐块 yield；反代请求超时阶梯——上传/执行类端点单独调大）。
|
||
- 报告登记端点 /api/bioflow/reports/register：BFF 侧落 doc_artifacts
|
||
（字段与 docgen 现有表对齐：job_id/标题/md/图清单——与 A3 report 模板协调），
|
||
落库后透传生信服务归一化响应。
|
||
- BIOFLOW_ENABLED 为假时路由不注册（main.py 按开关 include；三形态部署）。
|
||
|
||
约束：
|
||
- 不缓冲 SSE（TC-BIO-API-102 首帧延迟要求）。
|
||
- 反代失败统一 502 UPSTREAM_DOWN 信封（经 app.upstream.envelope）。
|
||
- 本文件是任务书列出的 BFF 三处最小改动之一。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, Depends, Request
|
||
from fastapi.responses import JSONResponse, Response, StreamingResponse
|
||
|
||
from ..config import get_settings
|
||
from ..db import Database, get_db
|
||
from ..upstream import envelope
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/bioflow", tags=["bioflow-proxy"])
|
||
|
||
# 反代超时阶梯（秒）：普通读 120；上传/报告登记等大体积端点单独调大
|
||
PROXY_READ_TIMEOUT = 120.0
|
||
PROXY_UPLOAD_TIMEOUT = 3600.0
|
||
PROXY_CONNECT_TIMEOUT = 5.0
|
||
|
||
_LONG_TIMEOUT_PATHS = ("/files", "/reports/register", "/archives", "/download")
|
||
|
||
|
||
def _proxy_timeout(path: str) -> float:
|
||
if any(seg in path for seg in _LONG_TIMEOUT_PATHS):
|
||
return PROXY_UPLOAD_TIMEOUT
|
||
return PROXY_READ_TIMEOUT
|
||
|
||
|
||
def _service_base() -> str:
|
||
return get_settings().bioflow_service_url.rstrip("/")
|
||
|
||
|
||
async def _proxy(request: Request, path: str) -> Any:
|
||
"""httpx 流式反代（SSE 透传不缓冲）；失败→502 信封。"""
|
||
import httpx
|
||
|
||
base = _service_base()
|
||
url = f"{base}/api/bioflow/{path.lstrip('/')}"
|
||
if request.url.query:
|
||
url = f"{url}?{request.url.query}"
|
||
body = await request.body()
|
||
headers = {k: v for k, v in request.headers.items()
|
||
if k.lower() not in ("host", "content-length", "connection")}
|
||
timeout = httpx.Timeout(connect=PROXY_CONNECT_TIMEOUT, read=_proxy_timeout(path),
|
||
write=PROXY_UPLOAD_TIMEOUT, pool=5.0)
|
||
# 手工管理 client 生命周期（不用 async with）：
|
||
# - 非 SSE：必须在 client 存活期内完成 aread()（BUG-R1-01 根因——此前
|
||
# aread 在 client 上下文退出后调用，uvloop 下连接已随 client 关闭 →
|
||
# httpx.ReadError → 502；大响应体首读未缓冲必现）。
|
||
# - SSE：client 须活到流消费完毕，故在 stream() finally 中一并 aclose。
|
||
client = httpx.AsyncClient(timeout=timeout)
|
||
try:
|
||
req = client.build_request(
|
||
request.method, url, content=body, headers=headers,
|
||
)
|
||
upstream = await client.send(req, stream=True)
|
||
except httpx.HTTPError as exc:
|
||
await client.aclose()
|
||
logger.warning("bioflow_proxy: 生信服务不可达 %s: %s", url, exc)
|
||
payload, status = envelope("UPSTREAM_DOWN", message="生信服务当前不可用，请稍后重试",
|
||
degraded=True, status=503)
|
||
return JSONResponse(status_code=status, content=payload)
|
||
|
||
content_type = upstream.headers.get("content-type", "")
|
||
if "text/event-stream" in content_type:
|
||
# SSE 透传：逐块 yield，绝不缓冲
|
||
async def stream():
|
||
try:
|
||
async for chunk in upstream.aiter_raw():
|
||
yield chunk
|
||
finally:
|
||
await upstream.aclose()
|
||
await client.aclose()
|
||
|
||
resp_headers = {
|
||
"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no",
|
||
"Connection": "keep-alive",
|
||
}
|
||
return StreamingResponse(stream(), status_code=upstream.status_code,
|
||
media_type="text/event-stream", headers=resp_headers)
|
||
|
||
# 非 SSE：client 存活期内完成读取（BUG-R1-01 修复）
|
||
try:
|
||
content = await upstream.aread()
|
||
except httpx.HTTPError:
|
||
await client.aclose()
|
||
payload, status = envelope("UPSTREAM_DOWN", message="生信服务响应中断",
|
||
degraded=True, status=502)
|
||
return JSONResponse(status_code=status, content=payload)
|
||
await upstream.aclose()
|
||
await client.aclose()
|
||
# 非 SSE：纯透传（content 是 bytes；不能 JSONResponse——bytes 序列化会 500）
|
||
media_type = content_type or "application/json"
|
||
return Response(status_code=upstream.status_code, content=content,
|
||
media_type=media_type)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 报告登记（BFF 侧落 doc_artifacts；FR-BIO-016/104 字段与 docgen 表对齐）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@router.post("/reports/register")
|
||
async def register_report(request: Request, db: Database = Depends(get_db)) -> Any:
|
||
"""生信报告 → doc_artifacts 行（BFF 拥有 docgen 表）。
|
||
|
||
先透传生信服务归一化响应（生信服务返回 {ok,data} 归一化载荷含图清单回填），
|
||
再落 doc_artifacts。落库失败不阻断反代响应（报告仍可直连下载；记日志）。
|
||
"""
|
||
try:
|
||
payload = await request.json()
|
||
except Exception:
|
||
payload, status = envelope("VALIDATION", message="请求体须为 JSON", status=422)
|
||
return JSONResponse(status_code=status, content=payload)
|
||
|
||
# 先透传生信服务归一化（含图清单回填）
|
||
proxied = await _proxy(request, "reports/register")
|
||
|
||
# BFF 落 doc_artifacts（尽力而为；字段对齐 docgen 表）
|
||
try:
|
||
import uuid
|
||
|
||
data = payload.get("data", payload) if isinstance(payload, dict) else {}
|
||
sid = str(data.get("session_id") or "")
|
||
user = getattr(request.state, "user", None)
|
||
if user and sid:
|
||
owner_row = await db.fetchone("SELECT owner_id FROM sessions WHERE id = ?", (sid,))
|
||
if owner_row is None or owner_row["owner_id"] != user["id"]:
|
||
return proxied # 归属不符：不落库，仍透传生信响应
|
||
artifact_id = uuid.uuid4().hex[:16]
|
||
await db.execute(
|
||
"""INSERT INTO doc_artifacts (id, session_id, status, title, template_name,
|
||
md_path, docx_path, pdf_path, error, created_at)
|
||
VALUES (?, ?, 'done', ?, 'bioflow_report', ?, NULL, NULL, NULL, ?)""",
|
||
(
|
||
artifact_id,
|
||
str(data.get("session_id") or ""),
|
||
str(data.get("title") or "生信分析报告"),
|
||
str(data.get("md_path") or ""),
|
||
_now_iso(),
|
||
),
|
||
)
|
||
logger.info("bioflow_proxy: 生信报告登记 doc_artifacts 行 %s（job=%s）",
|
||
artifact_id, data.get("job_id"))
|
||
except Exception: # noqa: BLE001 - 落库失败不阻断（docgen 表可能未建）
|
||
logger.exception("bioflow_proxy: doc_artifacts 落库失败（报告仍直连可用）")
|
||
|
||
return proxied
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 通用反代路由（其余 /api/bioflow/* 全部流式透传）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@router.api_route("/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"])
|
||
async def proxy_all(path: str, request: Request) -> Any:
|
||
return await _proxy(request, path)
|
||
|
||
|
||
def _now_iso() -> str:
|
||
from datetime import datetime, timezone
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|