title: FastAPI与SQLAlchemy数据库集成
date: 2025/04/17 15:33:34
updated: 2025/04/17 15:33:34
author: cmdragon

excerpt:
FastAPI与SQLAlchemy的集成通过创建虚拟环境、安装依赖、配置数据库连接、定义数据模型和实现路由来完成。核心模块包括数据库引擎、会话工厂和声明性基类。数据模型通过SQLAlchemy定义，路由通过FastAPI实现，支持创建和查询用户。测试接口通过curl命令进行，常见报错包括表不存在、请求体验证错误和会话不可用。高级配置涉及连接池优化和单元测试。

categories:

后端开发
FastAPI

tags:

FastAPI
SQLAlchemy
数据库集成
依赖注入
数据模型
错误处理
单元测试

图片[1]-FastAPI与SQLAlchemy数据库集成-牛翰网

扫描二维码关注或者微信搜一搜：编程智域前端至全栈交流与成长

探索数千个预构建的 AI 应用，开启你的下一个伟大创意

1. FastAPI与SQLAlchemy同步数据库集成实战

1.1 项目初始化与依赖安装

在项目根目录执行以下命令创建虚拟环境并安装依赖：

python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate.bat  # Windows
pip install fastapi sqlalchemy uvicorn

1.2 数据库配置核心模块

创建database.py文件：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 数据库连接配置（使用SQLite示例）
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

# 创建数据库引擎（关闭SQLite连接池检查）
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}
)

# 会话工厂（关闭自动提交，启用自动刷新）
SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
)

# 声明性基类
Base = declarative_base()


def get_db():
    """数据库会话依赖生成器"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

1.3 数据模型定义

创建models.py文件：

from sqlalchemy import Column, Integer, String
from database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)

1.4 依赖注入与路由集成

在main.py中实现：

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from models import User
from database import get_db, engine
from pydantic import BaseModel

# 初始化数据库表结构
Base.metadata.create_all(bind=engine)

app = FastAPI()


# Pydantic请求模型
class UserCreate(BaseModel):
    name: str
    email: str


@app.post("/users/")
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)
    return db_user


@app.get("/users/{user_id}")
def read_user(
        user_id: int,
        db: Session = Depends(get_db)
):
    """获取用户详情"""
    user = db.query(User).filter(User.id == user_id).first()
    return user

1.5 运行与测试

启动服务：

uvicorn main:app --reload

测试接口：

# 创建用户
curl -X POST "http://localhost:8000/users/" \
-H "Content-Type: application/json" \
-d '{"name":"John Doe","email":"john@example.com"}'

# 查询用户
curl "http://localhost:8000/users/1"

1.6 课后Quiz

问题1： 当数据库查询返回None时，如何优化API响应？

答案解析： 推荐使用FastAPI的HTTPException返回404状态码：

from fastapi import HTTPException


@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

问题2： 如何实现数据库事务回滚？

答案解析： 在依赖注入中使用try-except块：

@app.post("/orders/")
def create_order(db: Session = Depends(get_db)):
    try:
        # 数据库操作
        db.commit()
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail=str(e))

1.7 常见报错处理

报错1： sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: users

解决方案：

确认已执行Base.metadata.create_all(bind=engine)
检查数据库文件路径是否正确
删除旧数据库文件重新生成

报错2： pydantic.error_wrappers.ValidationError: 1 validation error

解决方案：

检查请求体是否与Pydantic模型定义匹配
验证字段类型和约束条件
使用curl测试时添加-H "Content-Type: application/json"

报错3： RuntimeError: Session is not available

解决方案：

确认路由函数正确使用Depends(get_db)
检查数据库连接配置是否正确
确保没有在路由外直接调用get_db()

1.8 高级配置技巧

连接池优化

# 配置MySQL连接池示例
engine = create_engine(
    "mysql+pymysql://user:password@localhost/dbname",
    pool_size=20,
    max_overflow=0,
    pool_recycle=3600
)

请求生命周期示意图

客户端请求 -> 路由处理 -> 创建数据库会话 -> 业务处理 -> 提交事务 -> 关闭会话
                      │                      │
                      └── 异常时回滚事务 ────┘

单元测试配置

# 测试用例示例
from fastapi.testclient import TestClient


def test_create_user():
    client = TestClient(app)
    response = client.post("/users/", json={
        "name": "Test User",
        "email": "test@example.com"
    })
    assert response.status_code == 200
    assert "id" in response.json()

余下文章内容请点击跳转至个人博客页面或者扫码关注或者微信搜一搜：编程智域前端至全栈交流与成长，阅读完整的文章：FastAPI与SQLAlchemy数据库集成 | cmdragon’s Blog