参考官方文档：https://fastapi.tiangolo.com/zh/tutorial/bigger-applications/

基本示例

from fastapi import APIRouterrouter = APIRouter()@router.get("/users/", tags=["users"])
async def read_users():return [{"username": "Rick"}, {"username": "Morty"}]@router.get("/users/me", tags=["users"])
async def read_user_me():return {"username": "fakecurrentuser"}@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):return {"username": username}

如何和app结合

from fastapi import Depends, FastAPIfrom .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, usersapp = FastAPI(dependencies=[Depends(get_query_token)])app.include_router(users.router)
app.include_router(items.router)
app.include_router(admin.router,prefix="/admin",tags=["admin"],dependencies=[Depends(get_token_header)],responses={418: {"description": "I'm a teapot"}},
)@app.get("/")
async def root():return {"message": "Hello Bigger Applications!"}

实战：改造用户注册登录接口

有了上面的基础之后，我们就可以改造之前的登录注册案例了。

改造执行的代码是这样的：

import time
import fastzdp_login
from fastapi import FastAPI, Depends, HTTPException, status, Body, APIRouter
from sqlalchemy.orm import Session as SASession
from sqlmodel import SQLModel, Field, Session, create_engine, select
from typing import Optionalclass User(SQLModel, table=True):id: Optional[int] = Field(default=None, primary_key=True)username: str = Field(index=True)password: str# 创建数据库引擎
sqlite_url = "mysql+pymysql://root:zhangdapeng520@127.0.0.1:3306/zdppy_demo?charset=utf8mb4"
engine = create_engine(sqlite_url, echo=True)# 确保表存在
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)app = FastAPI()# 伪造一个密钥，实际使用时应该使用安全的方式存储
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"# 令牌有效期
ACCESS_TOKEN_EXPIRE_MINUTES = 30# 假设你有一个获取用户的函数
def get_user(db: SASession, username: str):return db.exec(select(User).where(User.username == username)).first()# 依赖项，为每个请求提供数据库会话
def get_db():db = Session(engine)try:yield dbfinally:db.close()@app.post("/register", status_code=status.HTTP_201_CREATED)
def register_user(username: str = Body(str, min_length=2, max_length=36),password: str = Body(str, min_length=6, max_length=128),db: SASession = Depends(get_db),
):# 检查用户名是否已存在user = db.exec(select(User).where(User.username == username)).first()if user:raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Username already exists")# 创建新用户new_user = User(username=username, password=fastzdp_login.hash_256(password))db.add(new_user)try:db.commit()db.refresh(new_user)except Exception as e:print(e)db.rollback()raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Operation failed")return {"message": "User registered successfully", "user_id": new_user.id}# 登录接口
@app.post("/login", status_code=status.HTTP_200_OK)
async def login_for_access_token(username: str = Body(str, min_length=2, max_length=36),password: str = Body(str, min_length=6, max_length=128),db: SASession = Depends(get_db),
):user = get_user(db, username)if not user:raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")if not fastzdp_login.verify_256(password, user.password):raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password")data = {"username": user.username, "id": user.id, "time": time.time(), "expired": ACCESS_TOKEN_EXPIRE_MINUTES * 60}access_token = fastzdp_login.get_jwt(data, SECRET_KEY, ALGORITHM)return {"access_token": access_token, "token_type": "bearer"}if __name__ == '__main__':import uvicornuvicorn.run(app, host='0.0.0.0', port=8000)

改造之后的代码是这样的：

import fastzdp_login
from fastapi import FastAPI
from sqlmodel import SQLModel, Session, create_engine# 创建数据库引擎
sqlite_url = "mysql+pymysql://root:zhangdapeng520@127.0.0.1:3306/zdppy_demo?charset=utf8mb4"
engine = create_engine(sqlite_url, echo=True)# 确保表存在
SQLModel.metadata.drop_all(engine)
SQLModel.metadata.create_all(engine)app = FastAPI()# 伪造一个密钥，实际使用时应该使用安全的方式存储
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"# 令牌有效期
ACCESS_TOKEN_EXPIRE_MINUTES = 30def get_db():db = Session(engine)try:yield dbfinally:db.close()app.include_router(fastzdp_login.get_user_router(get_db,SECRET_KEY,ALGORITHM,ACCESS_TOKEN_EXPIRE_MINUTES * 60
))if __name__ == '__main__':import uvicornuvicorn.run(app, host='0.0.0.0', port=8000)