FastAPI 与数据库交互示例

安装必要的包

首先，需要安装以下包：

pip install fastapi uvicorn sqlalchemy pydantic

完整代码示例

from fastapi import FastAPI, HTTPException, Depends, status
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel
from typing import List, Optional

# 创建 FastAPI 应用
app = FastAPI(title="任务管理API")

# 数据库配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./tasks.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 定义数据库模型
class TaskModel(Base):
    __tablename__ = "tasks"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String)
    completed = Column(Boolean, default=False)

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

# 定义 Pydantic 模型用于请求和响应
class TaskBase(BaseModel):
    title: str
    description: Optional[str] = None
    completed: bool = False

class TaskCreate(TaskBase):
    pass

class Task(TaskBase):
    id: int
    
    class Config:
        orm_mode = True

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

# API 路由
@app.post("/tasks/", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(task: TaskCreate, db: Session = Depends(get_db)):
    db_task = TaskModel(**task.dict())
    db.add(db_task)
    db.commit()
    db.refresh(db_task)
    return db_task

@app.get("/tasks/", response_model=List[Task])
def read_tasks(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    tasks = db.query(TaskModel).offset(skip).limit(limit).all()
    return tasks

@app.get("/tasks/{task_id}", response_model=Task)
def read_task(task_id: int, db: Session = Depends(get_db)):
    task = db.query(TaskModel).filter(TaskModel.id == task_id).first()
    if task is None:
        raise HTTPException(status_code=404, detail="任务未找到")
    return task

@app.put("/tasks/{task_id}", response_model=Task)
def update_task(task_id: int, task: TaskCreate, db: Session = Depends(get_db)):
    db_task = db.query(TaskModel).filter(TaskModel.id == task_id).first()
    if db_task is None:
        raise HTTPException(status_code=404, detail="任务未找到")
    
    for key, value in task.dict().items():
        setattr(db_task, key, value)
    
    db.commit()
    db.refresh(db_task)
    return db_task

@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int, db: Session = Depends(get_db)):
    db_task = db.query(TaskModel).filter(TaskModel.id == task_id).first()
    if db_task is None:
        raise HTTPException(status_code=404, detail="任务未找到")
    
    db.delete(db_task)
    db.commit()
    return None

# 如果直接运行此文件，启动服务器
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

运行应用

保存上面的代码后，可以通过以下命令运行应用：

uvicorn fastapi_sql:app --reload

或者直接运行 Python 文件：

python fastapi_sql.py

使用说明

1. 启动应用后，访问 http://127.0.0.1:8000/docs 可以看到自动生成的 Swagger UI 文档
2. 通过这个界面可以测试所有的 API 端点
   

API 端点说明

• POST /tasks/ - 创建新任务
• GET /tasks/ - 获取所有任务列表
• GET /tasks/{task_id} - 获取特定任务详情
• PUT /tasks/{task_id} - 更新特定任务
• DELETE /tasks/{task_id} - 删除特定任务

代码解析

1. 数据库设置：

   • 使用 SQLAlchemy ORM 连接 SQLite 数据库
   • 创建数据库模型 TaskModel 定义表结构

2. Pydantic 模型：

   • 使用 Pydantic 定义数据验证和序列化模型
   • TaskBase 包含基本字段
   • TaskCreate 用于创建请求
   • Task 包含 ID，用于响应

3. 依赖注入：

   • 使用 Depends 注入数据库会话
   • 确保会话在请求结束后正确关闭

4. CRUD 操作：

   • 实现了完整的增删改查功能
   • 包含适当的错误处理和状态码

这个例子展示了 FastAPI 与 SQLAlchemy 的集成，适用于大多数数据库交互场景。如需使用其他数据库（如 MySQL、PostgreSQL），只需更改连接字符串和相应的驱动程序。