Skip to content

Transactions

Database transactions are supported thanks to encode/databases which is used to issue async queries.

Basic usage

To use transactions use database.transaction as async context manager:

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

Note

Note that it has to be the same database that the one used in Model's ormar_config object.

To avoid passing database instance around in your code you can extract the instance from each Model. Database provided during declaration of ormar.Model is available through ormar_config.database and can be reached from both class and instance.

 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

You can also use .transaction() as a function decorator on any async function:

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

Transaction blocks are managed as task-local state. Nested transactions are fully supported, and are implemented using database savepoints.

Manual commits/ rollbacks

For a lower-level transaction API you can trigger it manually

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

Testing

Transactions can also be useful during testing when you can apply force rollback and you do not have to clean the data after each test.

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
            ...