taiyi
9b32afdbb4
覆盖所有模块： - api 层：17 个路由文件，每个接口标注用途、参数、返回值、权限 - services 层：18 个服务文件，每个方法标注作用、参数、返回值、调用方 - repositories 层：13 个仓储文件，每个方法标注查询逻辑和被调用方 - schemas 层：11 个请求/响应体文件，每个字段标注业务含义 - core 层：config、security、exceptions、responses、error_codes - models 层：19 个 ORM 模型类，每个表标注业务含义和关联关系 - scripts：bootstrap_data、smoke_check - migrations：env.py 和版本迁移文件 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
"""
|
||
认证路由模块
|
||
|
||
职责：
|
||
处理用户认证相关接口，URL 前缀为 /api/auth。
|
||
包括：登录、获取当前用户信息、登出、绑定微信 OpenID。
|
||
"""
|
||
import json
|
||
|
||
from fastapi import APIRouter, Depends, Header
|
||
from pydantic import BaseModel
|
||
from sqlalchemy.orm import Session
|
||
|
||
from backend.app.api.deps import get_auth_service, get_current_user
|
||
from backend.app.db import get_db_session
|
||
from backend.app.schemas.auth import LoginRequest
|
||
from backend.app.schemas.common import success_payload
|
||
from backend.app.services.auth_service import AuthService
|
||
from backend.app.services.wechat_notification_service import wechat_notification_service
|
||
|
||
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
||
|
||
|
||
class BindOpenIdRequest(BaseModel):
|
||
"""绑定微信 OpenID 请求体"""
|
||
open_id: str
|
||
|
||
|
||
@router.post("/login")
|
||
def login(
|
||
payload: LoginRequest, # 登录请求体，包含用户名、密码、角色类型
|
||
auth_service: AuthService = Depends(get_auth_service), # 注入认证服务
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
) -> dict:
|
||
"""用户登录接口
|
||
|
||
用途：验证用户名和密码，返回 JWT Token。
|
||
请求参数：LoginRequest（username、password、role_type）。
|
||
返回值：登录结果，包含 token 等认证信息。
|
||
权限要求：无需认证（公开接口）。
|
||
"""
|
||
return success_payload(auth_service.login(payload.username, payload.password, payload.role_type, session))
|
||
|
||
|
||
@router.get("/me")
|
||
def me(
|
||
current_user: dict = Depends(get_current_user), # 从 JWT Token 解析当前用户
|
||
) -> dict:
|
||
"""获取当前登录用户信息
|
||
|
||
用途：返回当前已登录用户的详细信息（角色、权限等）。
|
||
请求参数：无（通过请求头 Authorization 传递 Token）。
|
||
返回值：当前用户信息字典。
|
||
权限要求：必须已登录（携带有效 Token）。
|
||
"""
|
||
return success_payload(current_user)
|
||
|
||
|
||
@router.post("/logout")
|
||
def logout(
|
||
authorization: str | None = Header(default=None), # 请求头中的 Authorization 字段
|
||
auth_service: AuthService = Depends(get_auth_service), # 注入认证服务
|
||
) -> dict:
|
||
"""用户登出接口
|
||
|
||
用途：使当前 Token 失效，完成登出。
|
||
请求参数：通过请求头 Authorization 传递 Bearer Token。
|
||
返回值：登出结果。
|
||
权限要求：无需严格校验，但需传递 Token。
|
||
"""
|
||
token = (authorization or "").removeprefix("Bearer").strip()
|
||
return success_payload(auth_service.logout(token))
|
||
|
||
|
||
@router.post("/bind-openid")
|
||
def bind_openid(
|
||
payload: BindOpenIdRequest, # 绑定请求体，包含微信 OpenID
|
||
session: Session = Depends(get_db_session), # 注入数据库会话
|
||
current_user: dict = Depends(get_current_user), # 当前登录用户
|
||
) -> dict:
|
||
"""绑定微信 OpenID 接口
|
||
|
||
用途：将当前用户的账号与微信 OpenID 绑定，用于微信消息通知。
|
||
请求参数：BindOpenIdRequest（open_id - 微信用户的 OpenID）。
|
||
返回值：绑定结果 {"success": true}。
|
||
权限要求：必须已登录。
|
||
"""
|
||
wechat_notification_service.bind_open_id(session, current_user["user_id"], payload.open_id)
|
||
return success_payload({"success": True})
|