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

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 运行与测试

启动服务：

1	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
)

请求生命周期示意图

1
2
3

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

单元测试配置

# 测试用例示例
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数据库集成

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

1.1 项目初始化与依赖安装

1.2 数据库配置核心模块

1.3 数据模型定义

1.4 依赖注入与路由集成

1.5 运行与测试

1.6 课后Quiz

1.7 常见报错处理

1.8 高级配置技巧

连接池优化

请求生命周期示意图

单元测试配置

往期文章归档：