Skip to content

交易

由于用于发出异步查询的编码/数据库，数据库事务得到支持。

基本用法

要使用事务，请使用database.transaction作为异步上下文管理器：

1
2
3
4
5
async with database.transaction():
    # everything called here will be one transaction
    await Model1().save()
    await Model2().save()
    ...

!!!note 请注意，它必须与模型的 ormar_config 对象中使用的数据库相同。

为了避免在代码中传递数据库实例，您可以从每个模型中提取实例。在 ormar.Model 声明期间提供的数据库可通过 ormar_config.database 获得，并且可以从类和实例访问。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import databases
import sqlalchemy
import ormar


base_ormar_config = OrmarConfig(
    metadata=sqlalchemy.MetaData(),
    database = databases.Database("sqlite:///"),
)


class Author(ormar.Model):
    ormar_config = base_ormar_config.copy()

    id: int = ormar.Integer(primary_key=True)
    name: str = ormar.String(max_length=255)

# database is accessible from class
database = Author.ormar_config.database

# as well as from instance
author = Author(name="Stephen King")
database = author.ormar_config.database

您还可以使用 .transaction() 作为任何异步函数的函数装饰器：

1
2
3
@database.transaction()
async def create_users(request):
    ...

事务块作为任务本地状态进行管理。完全支持嵌套事务，并使用数据库保存点来实现。

手动提交/回滚

对于较低级别的交易API，您可以手动触发它

1
2
3
4
5
6
7
8
transaction = await database.transaction()
try:
    await transaction.start()
    ...
except:
    await transaction.rollback()
else:
    await transaction.commit()

测试

当您可以应用强制回滚并且不必在每次测试后清理数据时，事务在测试期间也很有用。

1
2
3
4
5
6
@pytest.mark.asyncio
async def sample_test():
    async with database:
        async with database.transaction(force_rollback=True):
            # your test code here
            ...