Skip to content

deal

Database transactions are supported due to the encoding/database used to issue asynchronous queries.

basic usage

To use transactions, use database.transaction as the asynchronous 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 that it must be the same database used in the ormar_config object of the model.

To avoid passing database instances in code, you can extract instances from each model. The database provided during ormar.Model declaration is available through ormar_config.database and is accessible from classes and instances.

 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 for any asynchronous 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 implemented using database savepoints.

Manual commit/rollback

For lower-level trading APIs, 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()

test

Transactions are also useful during testing when you can apply forced rollbacks and don't have to clean up 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
            ...