129 lines
4.1 KiB
Python
129 lines
4.1 KiB
Python
"""认证路由：注册、登录、登出、用户信息。"""
|
||
|
||
from fastapi import APIRouter, Depends, Response
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.api.deps import get_current_user, get_db
|
||
from app.core.session import get_cookie_params
|
||
from app.models.user import User
|
||
from app.schemas.user import (
|
||
LoginRequest,
|
||
MeResponse,
|
||
RegisterRequest,
|
||
StorageInfoResponse,
|
||
UpdateMeRequest,
|
||
UserResponse,
|
||
)
|
||
from app.services.auth_service import AuthService
|
||
|
||
router = APIRouter(prefix="/auth", tags=["auth"])
|
||
|
||
|
||
@router.post("/register", response_model=UserResponse, status_code=201)
|
||
def register(body: RegisterRequest, response: Response, db: Session = Depends(get_db)) -> UserResponse:
|
||
"""注册新用户并自动登录。"""
|
||
auth_service = AuthService(db)
|
||
user, token = auth_service.register(body.username, body.email, body.password)
|
||
response.set_cookie(value=token, **get_cookie_params())
|
||
return _user_response(user)
|
||
|
||
|
||
@router.post("/login", response_model=UserResponse)
|
||
def login(body: LoginRequest, response: Response, db: Session = Depends(get_db)) -> UserResponse:
|
||
"""登录。"""
|
||
auth_service = AuthService(db)
|
||
user, token = auth_service.login(body.username_or_email, body.password)
|
||
response.set_cookie(value=token, **get_cookie_params())
|
||
return _user_response(user)
|
||
|
||
|
||
@router.post("/logout", status_code=204)
|
||
def logout(
|
||
response: Response,
|
||
user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> None:
|
||
"""登出：删除 session 并清除 Cookie。"""
|
||
from app.core.session import SESSION_COOKIE_NAME
|
||
|
||
token = "" # 不需要实际 token，logout 内部通过 user_id 找 session
|
||
auth_service = AuthService(db)
|
||
# 直接删除所有该用户的 session（MVP 简化：单设备）
|
||
from app.core.session import _store
|
||
|
||
to_delete = [t for t, e in _store.items() if e.user_id == user.id]
|
||
for t in to_delete:
|
||
auth_service.logout(t)
|
||
|
||
response.delete_cookie(SESSION_COOKIE_NAME, path="/")
|
||
return None
|
||
|
||
|
||
@router.get("/me", response_model=MeResponse)
|
||
def get_me(user: User = Depends(get_current_user)) -> MeResponse:
|
||
"""获取当前用户信息。"""
|
||
from app.core.config import get_settings
|
||
|
||
settings = get_settings()
|
||
return MeResponse(
|
||
id=user.id,
|
||
username=user.username,
|
||
email=user.email,
|
||
status=user.status,
|
||
storage_used=user.storage_used,
|
||
storage_quota=settings.default_storage_quota,
|
||
created_at=user.created_at,
|
||
)
|
||
|
||
|
||
@router.patch("/me", response_model=MeResponse)
|
||
def update_me(
|
||
body: UpdateMeRequest,
|
||
user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> MeResponse:
|
||
"""修改密码。"""
|
||
auth_service = AuthService(db)
|
||
if body.password is not None:
|
||
auth_service.change_password(user, body.password)
|
||
from app.core.config import get_settings
|
||
|
||
settings = get_settings()
|
||
return MeResponse(
|
||
id=user.id,
|
||
username=user.username,
|
||
email=user.email,
|
||
status=user.status,
|
||
storage_used=user.storage_used,
|
||
storage_quota=settings.default_storage_quota,
|
||
created_at=user.created_at,
|
||
)
|
||
|
||
|
||
@router.get("/storage", response_model=StorageInfoResponse)
|
||
def get_storage(user: User = Depends(get_current_user)) -> StorageInfoResponse:
|
||
"""获取存储用量信息。"""
|
||
from app.core.config import get_settings
|
||
|
||
settings = get_settings()
|
||
return StorageInfoResponse(
|
||
storage_used=user.storage_used,
|
||
storage_quota=settings.default_storage_quota,
|
||
storage_used_mb=round(user.storage_used / (1024 * 1024), 2),
|
||
storage_quota_mb=round(settings.default_storage_quota / (1024 * 1024), 2),
|
||
)
|
||
|
||
|
||
def _user_response(user: User) -> UserResponse:
|
||
from app.core.config import get_settings
|
||
|
||
settings = get_settings()
|
||
return UserResponse(
|
||
id=user.id,
|
||
username=user.username,
|
||
email=user.email,
|
||
status=user.status,
|
||
storage_used=user.storage_used,
|
||
storage_quota=settings.default_storage_quota,
|
||
created_at=user.created_at,
|
||
) |