55 lines
2.1 KiB
Python
55 lines
2.1 KiB
Python
"""认证路由：注册 / 登录 / 当前用户。"""
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..core.deps import get_current_user
|
|
from ..core.response import ok
|
|
from ..core.security import create_access_token, hash_password, verify_password
|
|
from ..database import get_db
|
|
from ..models import User
|
|
from ..schemas.auth import LoginRequest, RegisterRequest, TokenOut, UserOut
|
|
|
|
router = APIRouter(prefix="/auth", tags=["认证"])
|
|
|
|
|
|
def _user_out(u: User) -> UserOut:
|
|
return UserOut(
|
|
id=u.id, username=u.username, email=u.email, display_name=u.display_name,
|
|
avatar=u.avatar, role=u.role, created_at=u.created_at.isoformat(),
|
|
)
|
|
|
|
|
|
@router.post("/register")
|
|
def register(body: RegisterRequest, db: Session = Depends(get_db)):
|
|
if db.query(User).filter(User.username == body.username).first():
|
|
raise HTTPException(status_code=400, detail="用户名已被占用")
|
|
if db.query(User).filter(User.email == body.email).first():
|
|
raise HTTPException(status_code=400, detail="邮箱已被注册")
|
|
user = User(
|
|
username=body.username,
|
|
email=body.email,
|
|
hashed_password=hash_password(body.password),
|
|
display_name=body.display_name or body.username,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
token = create_access_token(user.id, user.username, user.role)
|
|
return ok(TokenOut(access_token=token, user=_user_out(user)).model_dump(), "注册成功")
|
|
|
|
|
|
@router.post("/login")
|
|
def login(body: LoginRequest, db: Session = Depends(get_db)):
|
|
user = db.query(User).filter(User.username == body.username).first()
|
|
if not user or not verify_password(body.password, user.hashed_password):
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
if not user.is_active:
|
|
raise HTTPException(status_code=403, detail="账号已禁用")
|
|
token = create_access_token(user.id, user.username, user.role)
|
|
return ok(TokenOut(access_token=token, user=_user_out(user)).model_dump(), "登录成功")
|
|
|
|
|
|
@router.get("/me")
|
|
def me(user: User = Depends(get_current_user)):
|
|
return ok(_user_out(user).model_dump())
|