fastapi搭建平台实战教程二：快速实现用户注册和登录

from fastapi import FastAPI
from pydantic import BaseModel

import models
from db import engine, SessionLocal
from fastapi.responses import JSONResponse


models.Base.metadata.create_all(bind=engine)

app = FastAPI()


class RegisterForm(BaseModel):
    username: str
    password: str

@app.post("/register")
def register(user: RegisterForm):
    if SessionLocal().query(models.User).filter_by(username=user.username).count() > 0:
        return JSONResponse({
            "code": 400,
            "msg": "用户已存在"
        })
    db_user = models.User(username=user.username, password=user.password)
    with SessionLocal.begin() as session:
        session.add(db_user)
    return JSONResponse({
            "code": 0,
            "msg": "注册成功"
        })

class LoginForm(BaseModel):
    username: str
    password: str

@app.post("/login")
def login(user:LoginForm):
    user = SessionLocal().query(models.User).filter_by(username = user.username).first()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    response = JSONResponse({
            "code": 0,
            "msg": "登录成功"
        })
    return response

from jose import jwt
from passlib.context import CryptContext

SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict):
    to_encode = data.copy()
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

@app.post("/register")
def register(user: RegisterForm):
    if SessionLocal().query(models.User).filter_by(username=user.username).count() > 0:
        return JSONResponse({
            "code": 400,
            "msg": "用户已存在"
        })
    db_user = models.User(username=user.username, password=get_password_hash(user.password))
    with SessionLocal.begin() as session:
        session.add(db_user)
    return JSONResponse({
            "code": 0,
            "msg": "注册成功"
        })

class LoginForm(BaseModel):
    username: str
    password: str

@app.post("/login")
def login(form:LoginForm):
    user = SessionLocal().query(models.User).filter_by(username = form.username).first()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if not verify_password(form.password, user.password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user.username}
    )
    response = JSONResponse({
            "code": 0,
            "msg": "登录成功",
            "access_token": access_token
        })
    return response

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    username: str = payload.get("sub")
    user = SessionLocal().query(models.User).filter_by(username = username).first()
    return user

class UserRes(BaseModel):
    username: str
    nickname: str | None = None
    email: str | None = None
    isadmin: bool | None = False

@app.get("/users/info", response_model=UserRes)
async def read_users_me(current_user: UserRes = Depends(get_current_user)):
    return current_user