mirror of
https://github.com/shy3130/tick-stock-panel.git
synced 2026-11-02 10:26:13 +00:00
shy3130
f20772d7b7
二开闭环补齐三块: 数据进得来 (写入)、事件推得出 (SSE)、契约有人守 (快照+CI)。
- write:ext scope (第六档): POST /api/ext-data/{id}/ingest 对 Token 开放,
复用管理端同一校验/落盘/视图刷新路径; 结构配置/上传/拉取仍不开放,
外部只能写行数据, 不能改写表结构
- SSE 短期票据 (V2 收尾): POST /api/events/ticket 换 60s 一次性票据 →
GET /api/events?ticket=… 订阅事件流; 票据只继承 Token scope 不放大权限;
进程内事件总线 (services/events.py, 慢消费者丢旧保新), 告警落盘点已挂
(alert_store.append/append_many → bus), 事件按 scope 过滤双向验证
- 契约守护: test_openapi_contract.py 冻结 tier=a 开放面 (61 端点),
增删须显式更新快照; 新增 ci.yml (后端全量测试 + 前端构建) 把契约
变化拦在 PR 上
- examples/open-api: 四个零依赖可运行示例 (行情/写入/回测/事件流) + README
- 菜单设置架构标注: 每页核心/扩展徽标 + 悬停原因 + 图例 (按 open-platform-plan
§2 依赖方向分类, 核心 8 页/扩展 11 页); 顺带补上 MenuSettings 缺失的 模拟盘
- 测试: 网关 write矩阵/票据语义 + 票据生命周期/总线/SSE帧/告警联动 + 契约快照,
全量 2408 passed / 102s; 线上实测: ingest 200 / 票据一次性 / SSE hello+
alert 推送 / scope 过滤双向 / 契约 61 路径
82 lines
3.1 KiB
Python
82 lines
3.1 KiB
Python
"""开放事件流 — SSE 短期票据 + /api/events 订阅端点。
|
|
|
|
EventSource 规范不能携带 Authorization 头, Token 调用方走两步:
|
|
POST /api/events/ticket (Bearer, 任意 scope) → 一次性票据 (60s)
|
|
GET /api/events?ticket=… (EventSource) → 按 scope 过滤的事件流
|
|
|
|
GET /api/events 在访问中间件白名单内 (票据即凭证, 不走密码会话);
|
|
票据校验失败立即 401, 成功则流式推送: hello 帧 → 事件 + 15s 心跳。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import queue
|
|
import time
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from fastapi.responses import JSONResponse, StreamingResponse
|
|
|
|
from app.services import event_tickets
|
|
from app.services.events import bus, sse_format
|
|
|
|
router = APIRouter(prefix="/api/events", tags=["events"])
|
|
|
|
_HEARTBEAT_S = 15.0
|
|
_POLL_S = 0.5
|
|
|
|
|
|
@router.post("/ticket")
|
|
def issue_ticket(request: Request):
|
|
"""Bearer Token (任意 scope) → 一次性 SSE 票据。"""
|
|
# 认证与限流已在网关中间件完成 (scope 规则 "*"); 端点内再验一次
|
|
# 明文是为了拿到 Token 记录、让票据继承 scope (不放大权限)。
|
|
authz = request.headers.get("authorization", "")
|
|
plaintext = authz[len("Bearer "):].strip() if authz.startswith("Bearer ") else ""
|
|
|
|
from app.services import api_tokens
|
|
|
|
record = api_tokens.verify_token(request.app.state.repo.store.data_dir, plaintext)
|
|
if record is None:
|
|
raise HTTPException(status_code=401, detail="API Token 无效或已吊销")
|
|
try:
|
|
ticket, ttl = event_tickets.issue(record.get("scopes", []))
|
|
except RuntimeError as e:
|
|
raise HTTPException(status_code=429, detail=str(e)) from e
|
|
return {"ticket": ticket, "expires_in": int(ttl), "stream": "/api/events?ticket=" + ticket}
|
|
|
|
|
|
@router.get("")
|
|
async def event_stream(ticket: str = ""):
|
|
"""SSE 事件流; 票据一次性, 重连需 POST /ticket 换新。"""
|
|
scopes = event_tickets.consume(ticket) if ticket else None
|
|
if scopes is None:
|
|
return JSONResponse(status_code=401, content={"detail": "票据无效、已使用或过期"})
|
|
|
|
q = bus.subscribe()
|
|
|
|
async def gen():
|
|
allowed = ("*", *scopes)
|
|
try:
|
|
yield sse_format({"type": "hello", "scope": "*", "data": {"scopes": scopes}, "ts": time.time()})
|
|
last_send = time.monotonic()
|
|
while True:
|
|
try:
|
|
event = q.get_nowait()
|
|
except queue.Empty:
|
|
if time.monotonic() - last_send >= _HEARTBEAT_S:
|
|
yield ": ping\n\n" # SSE 注释帧 — 保活, 不产生 EventSource 事件
|
|
last_send = time.monotonic()
|
|
await asyncio.sleep(_POLL_S)
|
|
continue
|
|
if event["scope"] in allowed:
|
|
yield sse_format(event)
|
|
last_send = time.monotonic()
|
|
finally:
|
|
bus.unsubscribe(q)
|
|
|
|
return StreamingResponse(
|
|
gen(),
|
|
media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
|
)
|