验证单个 Token 并返回更新后的行
(request: Request, token_id: int)
| 987 | |
| 988 | @router.post("/tokens/validate-single/{token_id}") |
| 989 | async def validate_single_token(request: Request, token_id: int): |
| 990 | """验证单个 Token 并返回更新后的行""" |
| 991 | from app.services.token_dao import get_token_dao |
| 992 | from app.utils.token_pool import get_token_pool |
| 993 | |
| 994 | dao = get_token_dao() |
| 995 | |
| 996 | # 验证 Token |
| 997 | await dao.validate_and_update_token(token_id) |
| 998 | |
| 999 | pool = get_token_pool() |
| 1000 | if pool: |
| 1001 | await pool.sync_from_database(DEFAULT_TOKEN_NAMESPACE) |
| 1002 | |
| 1003 | # 获取更新后的 Token 信息 |
| 1004 | async with dao.get_connection() as conn: |
| 1005 | cursor = await conn.execute(""" |
| 1006 | SELECT t.*, ts.total_requests, ts.successful_requests, ts.failed_requests, |
| 1007 | ts.last_success_time, ts.last_failure_time |
| 1008 | FROM tokens t |
| 1009 | LEFT JOIN token_stats ts ON t.id = ts.token_id |
| 1010 | WHERE t.id = ? |
| 1011 | """, (token_id,)) |
| 1012 | row = await cursor.fetchone() |
| 1013 | |
| 1014 | if row: |
| 1015 | # 返回更新后的单行 HTML |
| 1016 | token = dict(row) |
| 1017 | context = { |
| 1018 | "request": request, |
| 1019 | "token": token, |
| 1020 | } |
| 1021 | # 使用单行模板渲染 |
| 1022 | return templates.TemplateResponse("components/token_row.html", context) |
| 1023 | else: |
| 1024 | return HTMLResponse("") |
| 1025 | |
| 1026 | |
| 1027 | @router.post("/tokens/health-check") |
nothing calls this directly
no test coverage detected