| 12345678910111213141516171819202122232425262728293031323334 |
- """认证路由:用户注册、OAuth2 登录并返回 JWT。"""
- from fastapi import APIRouter, Depends, HTTPException, status
- from fastapi.security import OAuth2PasswordRequestForm
- from sqlmodel import Session, select
- from app.core.security import create_access_token, hash_password, verify_password
- from app.db import get_session
- from app.models import User
- from app.schemas import RegisterRequest, TokenResponse, UserPublic
- #创建路由
- router = APIRouter(prefix="/auth", tags=["auth"])
- #用户注册接口
- @router.post("/register", response_model=UserPublic)
- def register(payload: RegisterRequest, session: Session = Depends(get_session)):
- existing = session.exec(select(User).where(User.username == payload.username)).first()
- if existing:
- raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Username already exists")
- #创建用户
- user = User(username=payload.username, password_hash=hash_password(payload.password), role=payload.role)
- session.add(user)
- session.commit()
- session.refresh(user)
- return user
- #用户登录接口
- @router.post("/token", response_model=TokenResponse)
- def login(form: OAuth2PasswordRequestForm = Depends(), session: Session = Depends(get_session)):
- user = session.exec(select(User).where(User.username == form.username)).first()
- if not user or not verify_password(form.password, user.password_hash):
- raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password")
- token = create_access_token(subject=str(user.id), extra={"role": user.role})
- return TokenResponse(access_token=token)
|
