FastAPI 如何管理事务（session和多次提交）

在本文中，我们将介绍如何在 FastAPI 框架中使用 Sqlalchemy 来管理事务。事务是数据库中用于保持数据一致性的重要概念之一。在进行复杂的数据操作时，事务可以确保所有操作要么全部成功，要么全部失败。

什么是事务

事务是指一系列数据库操作的集合，这些操作要么全部成功，要么全部失败。事务应该具备以下四个特性，通常被简称为 ACID 特性：

原子性（Atomicity）：事务是不可分割的基本操作单元，要么全部执行成功，要么全部执行失败。
一致性（Consistency）：事务开始之前和执行结束之后，数据库的完整性约束不会被破坏。
隔离性（Isolation）：并发执行的事务之间相互隔离，一个事务的中间结果对其他事务是不可见的。
持久性（Durability）：事务成功提交后，其修改将永久保存在数据库中。

事务的管理对于应用程序的性能和可靠性至关重要。下面我们将介绍在 FastAPI 中如何使用 Sqlalchemy 来管理事务。

使用 Sqlalchemy 进行事务管理

在 FastAPI 中，可以使用 Sqlalchemy 提供的 session 和 commit 方法来管理事务。session 可以看作是一个数据库连接的上下文（context），通过 session 可以执行数据库操作，并确保这些操作可以原子地提交或回滚。

下面是一个示例代码，演示了如何在 FastAPI 中使用 Sqlalchemy 进行事务管理：

from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 创建数据库连接
engine = create_engine("数据库连接字符串")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

# 获取数据库 session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 创建用户
@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(username=user.username, email=user.email)
    db.add(db_user)  # 添加用户到 session
    db.commit()  # 提交事务
    db.refresh(db_user)  # 刷新用户
    return db_user

在上面的示例中，我们通过 create_engine 创建了数据库连接，并使用 sessionmaker 创建了 SessionLocal 类。SessionLocal 类通过 yield 关键字创建了上下文管理器，确保在每次请求处理时都能获取到一个数据库的 session，然后通过 db.commit() 来提交事务。

多次提交事务

有时，在一个请求处理中需要对数据库进行多次操作，并希望这些操作分别处于一个事务中，那么如何实现呢？Sqlalchemy 提供了 nested 操作符和 subtransaction 上下文管理器来实现这个功能。

下面是一个示例代码，演示了如何在 FastAPI 中实现多次提交事务的功能：

from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text

# 创建数据库连接
engine = create_engine("数据库连接字符串")
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

# 获取数据库 session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 在一个事务中进行多次提交
@app.post("/transactions/")
def process_transactions(transaction: TransactionSchema, db: Session = Depends(get_db)):
    with db.begin_nested():  # 开始一个嵌套事务
        db.execute(text("INSERT INTO transactions (amount) VALUES (:amount)"), {"amount": transaction.amount})
    # 其他数据库操作...
    with db.begin_nested():  # 开始另一个嵌套事务
        db.execute(text("UPDATE accounts SET balance = balance + :amount WHERE id = :account_id"),
                   {"amount": transaction.amount, "account_id": transaction.account_id})

    db.commit()  # 提交最外层事务
    return {"message": "Transactions processed successfully!"}

在上面的示例中，我们首先使用 with db.begin_nested(): 开始一个嵌套事务。在嵌套事务中，我们可以执行多个数据库操作，这些操作将在内部事务中进行。最后，通过 db.commit() 提交最外层事务。

总结

本文介绍了在 FastAPI 中使用 Sqlalchemy 来管理事务。我们了解了事务的概念和 ACID 特性，并通过示例代码展示了如何在 FastAPI 中使用 Sqlalchemy 进行事务管理。同时，还介绍了如何在一个请求处理中进行多次提交事务的方法。通过合理地管理事务，可以保证数据的一致性和可靠性，提高应用程序的性能。

希望本文对你理解 FastAPI 的事务管理有所帮助！