Transactions¶

Saffier uses SQLAlchemy 2.x Async transaction management directly, while keeping a familiar transaction API for application code.

You can see a transaction as atomic, which means, when you need to save everything or fail all.

Tip

Check more information about atomicity to get familiar with the concept.

There are three ways of using the transaction in your application:

As a decorator
As a context manager

The following explanations and examples will take in account the following:

Let us also assume we want to create a user and a profile for that user in a simple endpoint.

Danger

If you are trying to setup your connection within your application and have faced some errors related to a disconnected SQLAlchemy async engine, please see the connection section for more details and how to make it properly.

import saffier
from saffier import Database, Registry

database = Database("sqlite:///db.sqlite")
models = Registry(database=database)


class User(saffier.Model):
    """
    The User model to be created in the database as a table
    If no name is provided the in Meta class, it will generate
    a "users" table for you.
    """

    email = saffier.EmailField(unique=True, max_length=120)
    is_active = saffier.BooleanField(default=False)

    class Meta:
        registry = models


class Profile(saffier.Model):
    user = saffier.ForeignKey(User, on_delete=saffier.CASCADE)

    class Meta:
        registry = models

As a decorator¶

This is probably one of the less common ways of using transactions but still very useful if you want all of your endpoint to be atomic.

We want to create an endpoint where we save the user and the profile in one go. Since the author of Saffier is the same as Ravyn, it makes sense to use it as example.

You can use whatever you want, from Starlette to FastAPI. It is your choice.

from typing import TypedDict

from ravyn import Request, post
from models import Profile, User

from saffier import Database, Registry

# These settings should be placed somewhere
# Central where it can be accessed anywhere.
database = Database("sqlite:///db.sqlite")
models = Registry(database=database)


class UserIn(TypedDict):
    email: str


@post("/create", description="Creates a user and associates to a profile.")
@database.transaction()
async def create_user(data: UserIn, request: Request) -> None:
    # This database insert occurs within a transaction.
    # It will be rolled back by the `RuntimeError`.

    user = await User.query.create(email=data["email"], is_active=True)
    await Profile.query.create(user=user)
    raise RuntimeError()

As you can see, the whole endpoint is covered to work as one transaction. This cases are rare but still valid to be implemented.

As a context manager¶

This is probably the most common use-case for the majority of the applications where within a view or an operation, you will need to make some transactions that need atomocity.

from typing import TypedDict

from ravyn import Request, post
from models import Profile, User

from saffier import Database, Registry

# These settings should be placed somewhere
# Central where it can be accessed anywhere.
database = Database("sqlite:///db.sqlite")
models = Registry(database=database)


class UserIn(TypedDict):
    email: str


@post("/create", description="Creates a user and associates to a profile.")
async def create_user(data: UserIn, request: Request) -> None:
    # This database insert occurs within a transaction.
    # It will be rolled back by the `RuntimeError`.

    async with database.transaction():
        user = await User.query.create(email=data["email"], is_active=True)
        await Profile.query.create(user=user)
        raise RuntimeError()

Important notes¶

Saffier transactions run directly on SQLAlchemy's async transaction objects. Nested transaction blocks use SQLAlchemy savepoints, and force_rollback=True keeps test writes inside a rollback-only SQLAlchemy transaction for isolation.

If you need lower-level access, use the native SQLAlchemy async APIs exposed by database.engine, database.connection(), or database.session().