68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
from fastapi import APIRouter, Header
|
|
|
|
from app.dependencies import CurrentUser, DbSession, RedisClient
|
|
from app.schemas.auth import LoginRequest, RegisterRequest, TokenResponse, UserPublic
|
|
from app.schemas.common import ApiResponse
|
|
from app.services import auth as auth_service
|
|
|
|
router = APIRouter(prefix="/auth", tags=["认证"])
|
|
|
|
|
|
@router.post("/register", response_model=ApiResponse[TokenResponse])
|
|
async def register(
|
|
body: RegisterRequest,
|
|
db: DbSession,
|
|
redis: RedisClient,
|
|
) -> ApiResponse[TokenResponse]:
|
|
try:
|
|
result = await auth_service.register_user(
|
|
db,
|
|
username=body.username,
|
|
password=body.password,
|
|
)
|
|
except auth_service.AuthError as exc:
|
|
return ApiResponse(ok=False, message=exc.message)
|
|
|
|
await auth_service.store_session(redis, result.access_token, result.user.id)
|
|
return ApiResponse(ok=True, message="注册成功", data=result)
|
|
|
|
|
|
@router.post("/login", response_model=ApiResponse[TokenResponse])
|
|
async def login(
|
|
body: LoginRequest,
|
|
db: DbSession,
|
|
redis: RedisClient,
|
|
) -> ApiResponse[TokenResponse]:
|
|
try:
|
|
result = await auth_service.login_user(
|
|
db,
|
|
username=body.username,
|
|
password=body.password,
|
|
)
|
|
except auth_service.AuthError as exc:
|
|
return ApiResponse(ok=False, message=exc.message)
|
|
|
|
await auth_service.store_session(redis, result.access_token, result.user.id)
|
|
return ApiResponse(ok=True, message="登录成功", data=result)
|
|
|
|
|
|
@router.post("/logout", response_model=ApiResponse[None])
|
|
async def logout(
|
|
redis: RedisClient,
|
|
authorization: str | None = Header(default=None),
|
|
) -> ApiResponse[None]:
|
|
if authorization and authorization.lower().startswith("bearer "):
|
|
token = authorization[7:].strip()
|
|
if token:
|
|
await auth_service.revoke_session(redis, token)
|
|
return ApiResponse(ok=True, message="已退出登录")
|
|
|
|
|
|
@router.get("/me", response_model=ApiResponse[UserPublic])
|
|
async def me(current_user: CurrentUser) -> ApiResponse[UserPublic]:
|
|
return ApiResponse(
|
|
ok=True,
|
|
message="",
|
|
data=UserPublic.model_validate(current_user),
|
|
)
|