9.7 事务管理与回滚

FastAPI初学者教程：轻松掌握事务管理与回滚

FastAPI 教程

面向Python初学者的FastAPI入门教程，详细讲解事务管理与回滚的基础概念，通过生活化类比和动手实践示例，帮助零Web开发经验者快速上手并建立信心。

FastAPI初学者教程：事务管理与回滚

引言：为什么需要事务管理？

想象一下，你在网上购物时结账。如果只有一部分商品支付成功，而另一部分失败，你会感到困惑吧？这就是事务管理的核心：要么所有操作都成功，要么全部回滚（撤销）。在Web应用中，这确保了数据的一致性，避免因错误导致的部分更新。

第一部分：快速入门——建立第一个FastAPI应用

目标：运行一个简单的API，体验成功时刻

安装FastAPI：打开终端，运行命令安装所需库。
```
pip install fastapi uvicorn sqlalchemy
```

创建第一个应用：新建一个Python文件（如main.py），添加以下代码。

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def hello():
    return {"message": "欢迎来到FastAPI！你的第一个API已成功运行！"}

运行应用：在终端运行uvicorn main:app --reload，然后在浏览器访问http://127.0.0.1:8000/，看到返回的JSON消息，这就是你的成功时刻！

第二部分：数据库集成——连接SQLite数据库

目标：学会连接数据库并进行简单操作

配置数据库：使用SQLAlchemy连接SQLite，创建一个简单的表。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 创建数据库连接（SQLite文件数据库，简单易用）
engine = create_engine("sqlite:///./test.db", echo=True)
Base = declarative_base()
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 定义一个模型（如用户表）
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

# 创建表
Base.metadata.create_all(bind=engine)

测试数据库：运行应用后，数据库文件test.db会生成，你可以用SQLite浏览器查看，确认表已创建。

第三部分：事务管理与回滚——核心实践

目标：通过生活化类比，理解并实现事务

类比：事务就像银行转账——要么钱从一个账户转到另一个账户（成功），要么都不转（回滚），绝不会出现一半钱转了的情况。

动手实践：创建一个用户注册事务

假设用户注册时，需要同时创建用户记录和关联账户。我们使用SQLAlchemy的session管理事务。

定义事务操作：在FastAPI路由中处理。

from fastapi import FastAPI, HTTPException
from sqlalchemy.orm import Session
from .database import SessionLocal  # 假设导入之前定义的SessionLocal

app = FastAPI()

@app.post("/register/")
def register_user(name: str, email: str):
    # 获取数据库session
    db: Session = SessionLocal()
    try:
        # 开始事务（SQLAlchemy默认在session中管理事务）
        user = User(name=name, email=email)
        db.add(user)
        db.flush()  # 将操作暂存到数据库，但不提交

        # 模拟另一个操作：创建账户（为了演示事务，这里简单用另一个模型）
        # 假设有一个Account模型，但为简化，我们只用user示例
        # 如果这里出现错误，事务会回滚，用户不会被添加
        if email == "error@example.com":  # 故意添加错误场景
            raise HTTPException(status_code=400, detail="邮箱无效")

        # 提交事务：所有操作都成功
        db.commit()
        return {"success": "用户注册成功！事务已提交。"}
    except Exception as e:
        # 发生错误时，回滚事务
        db.rollback()
        raise HTTPException(status_code=500, detail=f"事务失败：{str(e)}")
    finally:
        db.close()

测试事务：
- 正常注册：发送POST请求到/register/?name=张三&email=zhangsan@example.com，应返回成功消息。
- 回滚场景：发送邮箱为error@example.com的请求，会看到错误消息，并检查数据库，用户记录不会被添加——这就是回滚的效果！

第四部分：扩展与总结

目标：巩固知识，鼓励进一步学习

成功回顾：你已经学会了如何创建FastAPI应用、连接数据库，并实现事务管理。每次运行代码看到预期结果，都是一次成功时刻！
生活化总结：事务就像生活中的“全有或全无”——比如订餐时，要么所有菜都送上，要么都不送，不会只送一半。
后续步骤：尝试添加更多模型，或使用异步数据库操作。记住，实践是最好的老师，多动手修改代码，探索FastAPI的强大功能。

附录：完整代码示例

为了方便，这里提供一个整合的main.py文件代码，你可以在一个文件中运行所有示例。

from fastapi import FastAPI, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, sessionmaker

# 数据库配置
engine = create_engine("sqlite:///./test.db", echo=True)
Base = declarative_base()
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 用户模型
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

# 创建表
Base.metadata.create_all(bind=engine)

app = FastAPI()

# 简单Hello API
@app.get("/")
def hello():
    return {"message": "FastAPI事务管理教程已启动！"}

# 用户注册事务示例
@app.post("/register/")
def register_user(name: str, email: str):
    db: Session = SessionLocal()
    try:
        user = User(name=name, email=email)
        db.add(user)
        db.flush()
        if email == "error@example.com":
            raise HTTPException(status_code=400, detail="邮箱无效，触发回滚！")
        db.commit()
        return {"success": "用户注册成功！事务已提交。"}
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=500, detail=f"事务失败：{str(e)}")
    finally:
        db.close()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

运行后，你可以通过浏览器或工具（如curl）测试API，体验事务管理的魔力。祝你学习愉快！

上一章 9.6 ORM 模式与数据模型转换

下一章 9.8 数据库连接池配置