FastAPI 集成 MySQL 和 Redis：模型与模式生成实践

在使用 FastAPI 构建应用程序时，通常会结合 MySQL 和 Redis 进行数据存储和缓存。为了生成模型（Model）和模式（Schema），你可以使用 SQLAlchemy 来与 MySQL 交互，并使用 Pydantic 来定义数据模式。以下是一个简单的示例，展示如何实现这一过程。

1. 安装依赖

首先，确保你已经安装了所需的依赖：

pip install fastapi uvicorn sqlalchemy pymysql redis pydantic

2. 创建 FastAPI 应用

创建一个 main.py 文件，并在其中编写 FastAPI 应用的代码。

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
import redis
from pydantic import BaseModel

# 创建 FastAPI 应用
app = FastAPI()

# MySQL 数据库配置
DATABASE_URL = "mysql+pymysql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Redis 配置
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# SQLAlchemy 模型
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), index=True)
    email = Column(String(50), unique=True, index=True)

# Pydantic 模式
class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str

    class Config:
        orm_mode = True

# 创建数据库表
Base.metadata.create_all(bind=engine)

# 依赖项：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 路由：创建用户
@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    # 将用户信息缓存到 Redis
    redis_client.set(f"user:{db_user.id}", db_user.email)
    return db_user

# 路由：获取用户
@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    # 尝试从 Redis 中获取用户信息
    cached_email = redis_client.get(f"user:{user_id}")
    if cached_email:
        return {"id": user_id, "name": "Cached User", "email": cached_email.decode('utf-8')}
    # 如果 Redis 中没有缓存，则从数据库中获取
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

# 运行应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

3. 代码解释

• SQLAlchemy 模型：User 类定义了数据库中的 users 表结构。
• Pydantic 模式：UserCreate 和 UserResponse 类用于定义请求和响应的数据模式。
• 数据库会话：get_db 函数用于获取数据库会话，并在请求结束后关闭会话。
• Redis 缓存：在创建用户时，将用户信息缓存到 Redis 中。在获取用户信息时，首先尝试从 Redis 中获取，如果不存在则从数据库中获取。

4. 运行应用

使用以下命令运行 FastAPI 应用：

uvicorn main:app --reload

5. 测试 API

你可以使用 curl 或 Postman 来测试 API：

• 创建用户：

  curl -X POST "http://127.0.0.1:8000/users/" -H "Content-Type: application/json" -d '{"name": "John Doe", "email": "john@example.com"}'
  

• 获取用户：

  curl "http://127.0.0.1:8000/users/1"