数据库事务回滚:FastAPI中的存档与读档大法

avatar
cmdragon 大乘
image image

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

探索数千个预构建的 AI 应用,开启你的下一个伟大创意https://tools.cmdragon.cn/

  1. 理解事务回滚机制

1.1 为什么需要事务回滚?
数据库事务回滚就像游戏中的存档机制,当执行某个任务失败时,可以回到任务开始前的状态。在Web开发中,当多个数据库操作需要保持原子性时(例如银行转账操作),事务回滚确保数据的一致性。

1.2 FastAPI中的事务实现
通过SQLAlchemy的session管理实现事务控制:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 创建数据库连接(使用SQLite示例)
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Pydantic模型
from pydantic import BaseModel


class UserCreate(BaseModel):
username: str
email: str


# 数据库模型
from sqlalchemy import Column, Integer, String
from database import Base


class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
username = Column(String(50), unique=True)
email = Column(String(100))

1.3 事务操作最佳实践
使用上下文管理器处理事务:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def create_user_with_profile(user_data: UserCreate):
db = SessionLocal()
try:
with db.begin():
# 创建用户
new_user = User(**user_data.dict())
db.add(new_user)

# 创建用户配置(假设需要原子操作)
db.execute("INSERT INTO profiles (user_id) VALUES (:user_id)",
{"user_id": new_user.id})

return new_user
except Exception as e:
db.rollback()
raise HTTPException(status_code=400, detail=str(e))
finally:
db.close()
  1. Alembic数据版本控制

2.1 迁移脚本生成
安装配置Alembic:

1
2
pip install alembic
alembic init migrations

修改alembic.ini:

1
sqlalchemy.url = sqlite:///./test.db

2.2 版本管理操作流程
生成新迁移脚本:

1
alembic revision -m "add phone_number column"

手动编辑生成的迁移文件:

1
2
3
4
5
6
7
8
def upgrade():
op.add_column('users',
Column('phone_number', String(20))
)


def downgrade():
op.drop_column('users', 'phone_number')

2.3 多环境迁移策略
在CI/CD中处理不同环境:

1
2
3
4
5
# 生产环境迁移
ALEMBIC_CONFIG=prod_alembic.ini alembic upgrade head

# 回滚到指定版本
alembic downgrade -1
  1. 综合应用案例

用户注册事务处理:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from fastapi import APIRouter, Depends

router = APIRouter()


@router.post("/register")
async def register_user(user: UserCreate):
db = SessionLocal()
try:
with db.begin():
# 检查用户名唯一性
if db.query(User).filter(User.username == user.username).first():
raise ValueError("Username already exists")

# 创建用户
new_user = User(**user.dict())
db.add(new_user)

# 创建关联记录
profile_data = {"user_id": new_user.id, "status": "active"}
db.execute(
"INSERT INTO profiles (user_id, status) VALUES (:user_id, :status)",
profile_data
)

return {"message": "Registration successful"}
except ValueError as ve:
db.rollback()
return {"error": str(ve)}
finally:
db.close()

课后Quiz:
Q: 当执行alembic downgrade命令时,系统实际执行什么操作?
A: 系统会按照迁移脚本的downgrade函数定义执行数据库结构回退操作,每个版本的downgrade会逆序执行。

常见报错解决:
问题:alembic.util.exc.CommandError: Can’t locate revision identified by ‘xxxx’
解决步骤:

  1. 检查migrations/versions目录是否存在对应版本的迁移文件
  2. 执行alembic history查看版本链
  3. 若存在断链,可删除alembic_version表后重新初始化

预防建议:
• 团队开发时统一管理迁移脚本
• 禁止手动修改数据库版本表
• 每次生成新迁移脚本后立即提交到版本控制系统

(完整示例代码需要安装fastapi, sqlalchemy, alembic等依赖,建议使用Python 3.8+环境,通过uvicorn main:app启动服务)

余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜:编程智域 前端至全栈交流与成长,阅读完整的文章:

往期文章归档: