FastAPI与SQLAlchemy数据库集成

怀念夏天 · 2025-4-17 14:42:03

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
数据库集成
依靠注入
数据模型
错误处理
单元测试

扫描二维码关注或者微信搜一搜：编程智域前端至全栈交流与发展
探索数千个预构建的 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数据库集成

本帖子中包含更多资源

0 个回复

快速回复

楼主热帖

标签云

浏览过的版块