当前位置：首页 > news >正文

FastAPI 与数据库交互示例

news 来源：原创 2025/5/3 6:42:37

安装必要的包

首先，需要安装以下包：

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: strdescription: Optional[str] = Nonecompleted: bool = Falseclass TaskCreate(TaskBase):passclass Task(TaskBase):id: intclass Config:orm_mode = True# 依赖项：获取数据库会话
def get_db():db = SessionLocal()try:yield dbfinally: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 uvicornuvicorn.run(app, host="127.0.0.1", port=8000)

运行应用

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

uvicorn fastapi_sql:app --reload

或者直接运行 Python 文件：

python fastapi_sql.py

使用说明

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

在这里插入图片描述

API 端点说明

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

代码解析

数据库设置：
- 使用 SQLAlchemy ORM 连接 SQLite 数据库
- 创建数据库模型 TaskModel 定义表结构
Pydantic 模型：
- 使用 Pydantic 定义数据验证和序列化模型
- TaskBase 包含基本字段
- TaskCreate 用于创建请求
- Task 包含 ID，用于响应
依赖注入：
- 使用 Depends 注入数据库会话
- 确保会话在请求结束后正确关闭
CRUD 操作：
- 实现了完整的增删改查功能
- 包含适当的错误处理和状态码