Transactions

Edgy using databases package allows also the use of transacations in a very familiar way for a lot of the users.

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 such as AssertationError: DatabaseBackend is not running, please see the connection section for more details and how to make it properly.

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

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 Edgy is the same as Esmerald, it makes sense to use it as example.

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

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()

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 atomacity. 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()

It is also possible to use 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)

Of course you can also access the database and start the transaction:

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)

Important notes

Edgy although running on the top of Databasez it varies in many aspects and offers features unprovided by sqlalchemy. For example the jdbc support or support for a mixed threading/async environment.

If you are interested in knowing more about the low-level APIs of databasez, check out or documentation.