| 123456789101112131415161718192021222324252627282930313233 |
- from fastapi import APIRouter, Depends, HTTPException
- from pydantic import BaseModel
- from sqlalchemy import select
- from sqlalchemy.ext.asyncio import AsyncSession
- from app.database import get_db
- from app.models.user import User
- import bcrypt
- import logging
- logger = logging.getLogger(__name__)
- router = APIRouter(prefix="/api/auth", tags=["auth"])
- class LoginRequest(BaseModel):
- username: str
- password: str
- @router.post("/login")
- async def login(payload: LoginRequest, db: AsyncSession = Depends(get_db)):
- result = await db.execute(
- select(User).where(User.username == payload.username)
- )
- user = result.scalar_one_or_none()
- if not user or not user.is_active:
- raise HTTPException(status_code=401, detail="用户名或密码错误")
- pw_hash = user.password_hash.encode("utf-8")
- if not bcrypt.checkpw(payload.password.encode("utf-8"), pw_hash):
- raise HTTPException(status_code=401, detail="用户名或密码错误")
- return {"message": "登录成功", "username": user.username}
|
