Python数据API事务怎么管理

本文目录导读：

1. 使用 SQLAlchemy（最常用）
2. 使用 Django ORM
3. 使用 psycopg2（原生 PostgreSQL）
4. 使用 asyncpg（异步 PostgreSQL）
5. 事务隔离级别控制
6. 最佳实践建议
7. 注意事项

在Python数据API中管理事务,主要取决于你所使用的数据库和ORM/驱动，以下是几种常见场景的管理方式：

使用 SQLAlchemy（最常用）

自动提交模式

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
engine = create_engine('postgresql://user:pass@localhost/db')
session = Session(engine)
# 自动提交示例
with session.begin():
    user = User(name="Alice")
    session.add(user)
    # 事务会自动提交或回滚

显式提交/回滚

session = Session(engine)
try:
    user1 = User(name="Bob")
    user2 = User(name="Charlie")
    session.add_all([user1, user2])
    session.commit()  # 提交事务
except Exception as e:
    session.rollback()  # 回滚事务
    raise e
finally:
    session.close()

使用 Django ORM

基于装饰器

from django.db import transaction
@transaction.atomic
def create_user_and_profile():
    user = User.objects.create(username='alice')
    Profile.objects.create(user=user, bio='Hello')
    # 如果任一操作失败，整个事务回滚

上下文管理器

def bank_transfer(from_account, to_account, amount):
    with transaction.atomic():
        from_account.balance -= amount
        from_account.save()
        to_account.balance += amount
        to_account.save()

使用 psycopg2（原生 PostgreSQL）

import psycopg2
conn = psycopg2.connect("dbname=test user=postgres")
try:
    conn.autocommit = False  # 关闭自动提交
    cur = conn.cursor()
    cur.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cur.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()  # 正常提交
except Exception as e:
    conn.rollback()  # 出错回滚
    raise e
finally:
    conn.close()

使用 asyncpg（异步 PostgreSQL）

import asyncpg
async def transfer_money():
    conn = await asyncpg.connect(user='user', password='pass', database='test')
    try:
        async with conn.transaction():
            await conn.execute('UPDATE accounts SET balance = balance - 100 WHERE id = 1')
            await conn.execute('UPDATE accounts SET balance = balance + 100 WHERE id = 2')
    finally:
        await conn.close()

事务隔离级别控制

from sqlalchemy import create_engine
from sqlalchemy.orm import Session
# 设置隔离级别
engine = create_engine(
    'postgresql://user:pass@localhost/db',
    isolation_level="REPEATABLE READ"  # 可选：READ COMMITTED, SERIALIZABLE
)
# 或在使用时设置
with engine.connect() as conn:
    conn.execution_options(isolation_level="SERIALIZABLE")
    # 执行事务操作

最佳实践建议

事务装饰器封装

from functools import wraps
from sqlalchemy.orm import Session
def transactional(func):
    @wraps(func)
    def wrapper(session: Session, *args, **kwargs):
        try:
            result = func(session, *args, **kwargs)
            session.commit()
            return result
        except Exception as e:
            session.rollback()
            raise e
    return wrapper
@transactional
def create_order(session, order_data, user_id):
    # 业务逻辑
    order = Order(**order_data)
    session.add(order)
    return order

嵌套事务处理

# SQLAlchemy 的嵌套事务
with session.begin():
    # 外层事务
    obj1 = MyModel(value=1)
    session.add(obj1)
    with session.begin_nested():  # savepoint
        obj2 = MyModel(value=2)
        session.add(obj2)
        # 内层回滚不会影响外层

注意事项

1. 事务尽量短：长时间占用连接资源
2. 合理设置隔离级别：权衡数据一致性与性能
3. 异常处理：确保所有路径都有清晰的错误处理
4. 连接池管理：避免连接泄漏
5. 分布式事务：考虑使用Saga模式或两阶段提交

选择哪种方式取决于你的项目框架、数据库类型和性能需求，对于Flask/FastAPI应用推荐SQLAlchemy+显式事务管理，Django项目使用其内置ORM即可。