6fc1abf2f5
CI/CD Pipeline / Unit Tests (pull_request) Failing after 7s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 12s
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 1m44s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Failing after 2m36s
- P2-1: 10处 except 审查改进(细化异常类型、添加日志) - P2-2: 22个路由函数类型注解补全 - P2-3: edit_plans.py 拆分为 4 个模块(CRUD/generation/ai/timeline) - P2-4: generate_video + generate_plan 大函数拆分 - P2-5: unified_render_service.py 拆分(1517→984行) - render_audio.py: 音频混音模块(RenderContext + mix/merge 函数) - render_subtitles.py: ASS 字幕生成模块 - P2-6: 6个未使用配置项删除确认 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
"""
|
|
Authentication dependency compatibility layer.
|
|
|
|
Canonical bearer-token parsing lives in app.auth. This module re-exports
|
|
common auth dependencies for backward compatibility.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from app.auth import AuthenticatedUser
|
|
from app.auth import get_current_user as get_authenticated_user
|
|
from app.dependencies import get_user_repository
|
|
from fastapi import Depends, HTTPException
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from packages.domain.entities import User
|
|
from packages.ports.user_repository import UserRepository
|
|
|
|
optional_bearer_scheme = HTTPBearer(auto_error=False)
|
|
|
|
|
|
async def get_current_user(
|
|
authenticated_user: AuthenticatedUser = Depends(get_authenticated_user),
|
|
) -> User:
|
|
return authenticated_user.user
|
|
|
|
|
|
async def get_current_user_optional(
|
|
credentials: HTTPAuthorizationCredentials | None = Depends(optional_bearer_scheme),
|
|
user_repository: UserRepository = Depends(get_user_repository),
|
|
) -> User | None:
|
|
if credentials is None:
|
|
return None
|
|
try:
|
|
authenticated_user = await get_authenticated_user(credentials, user_repository)
|
|
except HTTPException as exc:
|
|
if exc.status_code >= 500:
|
|
# 服务端错误不应被静默吞掉,记录日志
|
|
logger.error("可选认证遇到服务端错误,status=%s", exc.status_code, exc_info=True)
|
|
# 4xx 认证失败(如 token 无效、用户不存在)属于正常流程,返回 None
|
|
return None
|
|
return authenticated_user.user
|