Integration with Tortoise ORM.

[0x1618]

Hi, i'm wondering how to integrate falcon with Tortoise ORM. As far as i know, Tortoise ORM manages transactions on its own, but i'm not sure if it creates a transaction for each request (although i don't see why it would). I already managed to create process_startup and process_shutdown, and i think that to create a new transaction (where exceptions trigger a rollback), you need to use async with in_transaction() wherever you run database queries.

Here you can see an example. There’s one problem: entity_user is fetched in a separate transaction (which can lead to data mismatch), but it should be fetched within a single transaction using in_transaction(). As you can see, it's challenging to manage this without proper middleware. Without in_transaction(), if an exception occurs in verification(), the database will save progress up to that point, which could lead to bugs.

@before(authorize_user, is_logged=True, is_email_verified=False)
class VerifyEmailResource:
    async def on_get(self, req: Request, resp: Response):
        entity_user: UserType = req.context.entity_user # Got from authorize_user. (Used Tortoise ORM)

        async with in_transaction():
            await entity_user.verification().start_verification() # Tortoise ORM operations.

I'm thinking about creating a decorator to wrap on_get so it would run on_get within a async with in_transaction() block. However, i'm not sure how to do it properly to make it work well with Falcon and Tortoise ORM, also there may be a better solution.

Current middleware:

class TortoiseMiddleware:
    async def process_startup(self, scope, event):
        await Tortoise.init(config=TORTOISE_ORM)
        await Tortoise.generate_schemas()

    async def process_shutdown(self, scope, event):
        await Tortoise.close_connections()

I would really appreciate help with the integration. Also if you have a good solution for other ORM integration, please feel free to share it.

[0x1618]

Ok, so i created a decorator, create_transaction(), which wraps the entire route in a with block, but it feels a bit nested...

def create_transaction():
    def decorator(func):
        @wraps(func)
        async def wrapped(self, req, resp, *args, **kwargs):
            async with in_transaction() as connection:
                req.context.connection = connection

                return await func(self, req, resp, *args, **kwargs)

        return wrapped

    return decorator

class VerifyEmailResource:
    @create_transaction()
    @before(authorize_user, is_logged=True, is_email_verified=False) # uses req.context.connection
    async def on_get(self, req: Request, resp: Response):
        entity_user: UserType = req.context.entity_user

        await entity_user.verification().start_verification()

        resp.status = 200
        resp.media = {
            "title": VERIFICATION_STARTED_TITLE,
            "description": VERIFICATION_STARTED_MSG
        }

    @create_transaction()
    @before(authorize_user, is_logged=True, is_email_verified=False) # uses req.context.connection
    @before(
        validate_request,
        token={
            'type': str,
            'validators': {
                'exact_length': VERIFICATION_TOKEN_LENGTH
            }
        }
    )
    async def on_post(self, req: Request, resp: Response):
        entity_user: UserType = req.context.entity_user

        await entity_user.verification().end_verification(
            token=req.context.data['token']
        )

        resp.status = 200
        resp.media = {
            "title": VERIFICATION_ENDED_TITLE,
            "description": VERIFICATION_ENDED_MSG
        }

A much better option would be to do something like this under. However, i'm not sure how to implement the decorator properly to map all routes, like @before do.

@create_transaction()
@before(authorize_user, is_logged=True, is_email_verified=False) # uses req.context.connection
class VerifyEmailResource:
    async def on_get(self, req: Request, resp: Response):
        entity_user: UserType = req.context.entity_user

        await entity_user.verification().start_verification()

        resp.status = 200
        resp.media = {
            "title": VERIFICATION_STARTED_TITLE,
            "description": VERIFICATION_STARTED_MSG
        }


    @before(
        validate_request,
        token={
            'type': str,
            'validators': {
                'exact_length': VERIFICATION_TOKEN_LENGTH
            }
        }
    )
    async def on_post(self, req: Request, resp: Response):
        entity_user: UserType = req.context.entity_user

        await entity_user.verification().end_verification(
            token=req.context.data['token']
        )

        resp.status = 200
        resp.media = {
            "title": VERIFICATION_ENDED_TITLE,
            "description": VERIFICATION_ENDED_MSG
        }

[CaselIT]

Hi,

This looks like a discussion, so I will move it there.
Regarding your question, falcon has middleware methods that run before and after a responder method so you could use them to start a transaction then commit/rollback.

The method are documented here process_resource and process_response. See the docs here https://falcon.readthedocs.io/en/stable/api/middleware.html
You could do something like the following:

class Middleware:
    async def process_resource(
        self,
        req: Request,
        resp: Response,
        resource: object,
        params: dict[str, Any],
    ) -> None:
        req.context.conn = await new_transaction() # create a transaction
    async def process_response(
        self,
        req: Request,
        resp: Response,
        resource: object,
        req_succeeded: bool
    ) -> None:
        if req_succeded:
           await req.context.conn.commit()
        else:
           await req.context.conn.rollback()

[0x1618]

Thanks for the answer, but this does not provide session rollback in case of an error. You need to wrap your route in a decorator that uses await with in_transaction() as connection, (or wrap the content of the route). Also i researched that connections are strictly related to database connections, not transactions themselves, so my code in the initial question was flawed.

[CaselIT]

this does not provide session rollback in case of an error

if you have an error req_succeeded is False, so you can rollback the session in that case.
Note that I'm not sure if it's possible using tortoise

[0x1618]

Hi. I ended up with this solution. Tortoise ORM natively supports wrapping a function in a single transaction using the Atomic decorator. To ensure a request is in an independent transaction (and will roll back in case of an exception), you need to wrap it with tortoise.transactions.atomic.

from tortoise import Tortoise, fields
from tortoise.models import Model
from tortoise.transactions import atomic as req_in_transaction

class TestModel(Model):
    uuid = fields.UUIDField(primary_key=True)

class TortoiseMiddleware:
    async def process_startup(self, scope, event):
        await Tortoise.init(config=TORTOISE_ORM)
        await Tortoise.generate_schemas()

    async def process_shutdown(self, scope, event):
        await Tortoise.close_connections()

class TransactionTestResource:
    @req_in_transaction() # Request is in one transaction.
    async def on_post(self, req, resp):
        db_test = TestModel()

        await db_test.save()
        
        assert 2 == 3 # Because of the exception db_test won't be saved (session.rollback). Without atomic (req_in_transaction) decorator, db_test would be saved.

        resp.status = 200
        resp.media = {
            'message': 'Test'
        }

[vytas7]

Hi @0x1618,
Unfortunately I don't know much about Tortoise ORM, but you should be able to enter the context in question using middleware too, if you prefer to do it in a centralized way. You could use AsyncExitStack from the stdlib:

import contextlib
import logging

import falcon.asgi

logging.basicConfig(
    format='%(asctime)s [%(levelname)s] %(message)s', level=logging.INFO)


@contextlib.asynccontextmanager
async def in_transaction():
    logging.info('Entering async context...')
    try:
        yield 'Hello, World!'
    finally:
        logging.info('Finally leaving async context...')


class AsyncContextMiddleware:
    async def process_request(self, req, resp):
        req.context.exit_stack = contextlib.AsyncExitStack()

    async def process_resource(self, req, resp, resource, params):
        await req.context.exit_stack.enter_async_context(in_transaction())

    async def process_response(self, req, resp, resource, req_succeeded):
        await req.context.exit_stack.aclose()


class RequestLog:
    async def process_response(self, req, resp, resource, req_succeeded):
        logging.info(f'{req.method} {req.path} ==> HTTP {resp.status_code}')


class HelloResource:
    async def on_get(self, req, resp):
        resp.media = {'message': 'Hello, World!'}


app = falcon.asgi.App(middleware=[RequestLog(), AsyncContextMiddleware()])
app.add_route('/hello', HelloResource())

And then running it with Uvicorn:

2024-11-04 11:45:14,591 [INFO] Entering async context...
2024-11-04 11:45:14,591 [INFO] Finally leaving async context...
2024-11-04 11:45:14,591 [INFO] GET /hello ==> HTTP 200

If desired, you could also wrap the closing part with try... except, and handle the errors yourself:

    async def process_response(self, req, resp, resource, req_succeeded):
        try:
            await req.context.exit_stack.aclose()
        except Exception:
            logging.exception('Error closing exit stack')

If you don't do anything special, Falcon should render a generic 500 for you on exceptions both when entering and leaving the context.

[0x1618]

That’s clever. But is it uvicorn-safe? I mean, is it thread-safe? Doesn’t it "wrap" every code block it encounters after the middleware is executed? I think this could lead to issues. A decorator has the benefit of wrapping only the specified code block without affecting others.

[vytas7]

It shouldn't be conceptually different from wrapping your responder method with a decorator.
Re wrapping just specific methods, yes, middleware applies to all responder methods, so you could check the resource parameter if a database context is applicable. And yes, if you want to only "wrap" only a handful of specific responders, a direct decorator is better. If you want to wrap the majority, then it is better to DRY it with middleware.

And I'm not sure I understand the point re thread-safety. Is Tortoise ORM spawning background threads?