Transactions in Edgy

Edgy, leveraging the databasez package, provides robust transaction support that will feel familiar to many developers. Transactions ensure atomicity, meaning that a series of database operations either all succeed or all fail, maintaining data consistency.

Tip

For a deeper understanding of atomicity, refer to the Atomicity in Database Systems documentation.

Edgy offers three primary ways to manage transactions:

The following examples will use a scenario where we create a user and a profile for that user within a single endpoint.

Danger

If you encounter AssertionError: DatabaseBackend is not running, please consult the connection section for proper connection setup.

import edgy
from edgy import Database, Registry

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


class User(edgy.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: str = edgy.EmailField(unique=True, max_length=120)
    is_active: bool = edgy.BooleanField(default=False)

    class Meta:
        registry = models


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

    class Meta:
        registry = models

As a Decorator

Using transactions as decorators is less common but useful for ensuring entire endpoints are atomic.

Consider an Esmerald endpoint (but this can be any web framework) that creates a user and a profile in one atomic operation:

from esmerald import Request, post
from models import Profile, User
from pydantic import BaseModel, EmailStr

from edgy import Database, Registry

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


class UserIn(BaseModel):
    email: EmailStr


@post("/create", description="Creates a user and associates to a profile.")
@models.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()

In this case, the @transaction() decorator ensures that the entire endpoint function executes within a single transaction. This approach is suitable for cases where all operations within a function must be atomic.

As a Context Manager

Context managers are the most common way to manage transactions, especially when specific sections of code within a view or operation need to be atomic. It is recommended to use the model or queryset transaction method. This way the transaction of the right database is used.

from esmerald import Request, post
from models import Profile, User
from pydantic import BaseModel, EmailStr

from edgy 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(BaseModel):
    email: EmailStr


@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 User.transaction():
        user = await User.query.create(email=data.email, is_active=True)
        await Profile.query.create(user=user)
        raise RuntimeError()

Using the current active database of a QuerySet:

from esmerald import Request, post
from models import Profile, User
from pydantic import BaseModel, EmailStr

from edgy import Database, Registry

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


class UserIn(BaseModel):
    email: EmailStr


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

    queryset = User.query.using(database="another")

    async with queryset.transaction(force_rollback=True):
        user = await queryset.create(email=data.email, is_active=True)
        await Profile.query.using(database="another").create(user=user)

You can also access the database and start the transaction directly:

from esmerald import Request, post
from models import Profile, User
from pydantic import BaseModel, EmailStr

from edgy import Database, Registry

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


class UserIn(BaseModel):
    email: EmailStr


@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 a transaction with force_rollback active.

    queryset = User.query.using(database="another")

    async with queryset.database as database, database.transaction(force_rollback=True):
        user = await queryset.create(email=data.email, is_active=True)
        await Profile.query.using(database="another").create(user=user)

This ensures that the operations within the async with block are executed atomically. If any operation fails, all changes are rolled back.

Important Notes

Edgy, while built on top of Databasez, offers unique features beyond those provided by SQLAlchemy. These include JDBC support and compatibility with mixed threading/async environments.

For more information on the low-level APIs of Databasez, refer to the Databasez repository and its documentation.