SQLAlchemy Core events

[tng10]

• GINO version: 0.8.1
• Python version: 3.6.6
• asyncpg version: 0.18.2
• aiocontextvars version: 0.2.1
• PostgreSQL version: 10.6

Description

Hello, I would like to know if it is possible to use SQLAlchemy Core Events with Gino.

It would be really great to use DDLEvents

events.py

from sqlalchemy import event
from app.models import UserModel

def after_create_method(target, connection, **kw):
    print('some routine...')

event.listen(UserModel.__table__, "after_create", after_create_method)

create_user_example.py

async def create_user(validated_data):
    instance = await UserModel.create(**validated_data)
    return instance

I've managed to register a listener by passing table attribute, but unfortunately the function doesn't get invoked once the UserModel.create gets called. Can someone help me on that?

I've seen gino.Crud.CRUDModel.create is the class responsible for creating a record on the database. In case we are moving forward with this issue, is it possible to come up with SQLAlchemy events on this method/class?

Thanks in advance!

[fantix]

SQLAlchemy events are quite hacky in GINO, I'll reply a bit later.

[tng10]

Hey @fantix,

I've been working on it and that's true, it is a bit hacky. The idea is to provide access to the events through hooks on the gino/crud.py file. It would be great to have either mixins or composition in order to provide such access.

All I can say by now is that I already managed to get it working, I'll reply a bit later once I have a better understanding of this feature.

Thanks!

[tng10]

Hi,

I created a mixin and added it to CRUDModel class, this way we can take advantage of the SQLAlchemy Core Events. The code is a bit messy and can be improved for sure, but I would like to hear your opinion about it. What do you think?

mixins.py

import enum, asyncio
from sqlalchemy.event import registry

class DDLEvents(enum.Enum):
    AFTER_CREATE = 'after_create'
    AFTER_UPDATE = 'after_update'
    AFTER_DELETE = 'after_delete'

class GinoSQLAlchemyCoreEvents:
    events = registry._key_to_collection

    async def call_method(self, method_to_call, *args, **kwargs):
        if asyncio.iscoroutinefunction(method_to_call):
            await method_to_call(*args, **kwargs)
        else:
            method_to_call(*args, **kwargs)

    def filter_events(self, event_type, cls):
        filtered_events = {}
        for event_key, event_value in self.events.items():
            if event_key[0] == id(cls.__table__) and event_key[1] == event_type:
                filtered_events.update({event_key: event_value})
        return filtered_events

    def after_create_events(self, cls):
        return self.filter_events(DDLEvents.AFTER_CREATE.value, cls)

    def after_update_events(self, cls):
        return self.filter_events(DDLEvents.AFTER_UPDATE.value, cls)

    def after_delete_events(self, cls):
        return self.filter_events(DDLEvents.AFTER_DELETE.value, cls)

    async def dispatch_events(self, event_type, cls, bind, instance, row):
        events_to_dispatch = f'{event_type}_events'
        for event_key, event_value in getattr(self, events_to_dispatch)(cls).items():
            method_to_call = list(event_value.values())[0]()
            if method_to_call:
                args = (cls, bind, instance, row)
                kwargs = {}
                await self.call_method(method_to_call, *args, **kwargs)

    async def after_create(self, cls, bind, instance, row):
        await self.dispatch_events(DDLEvents.AFTER_CREATE.value, cls, bind, instance, row)

    async def after_update(self, cls, bind, instance, row):
        await self.dispatch_events(DDLEvents.AFTER_UPDATE.value, cls, bind, instance, row)

    async def after_delete(self, cls, bind, instance, row):
        await self.dispatch_events(DDLEvents.AFTER_DELETE.value, cls, bind, instance, row)

crud.py

...
from .mixins import GinoSQLAlchemyCoreEvents
...

...
class CRUDModel(Model, GinoSQLAlchemyCoreEvents):
...

_create ...
        if bind is None:
            bind = cls.__metadata__.bind
        row = await bind.first(q)

        for k, v in row.items():
            self.__values__[self._column_name_map.invert_get(k)] = v
        self.__profile__ = None

        await self.after_create(cls, bind, self, row)

        return self
...

apply ...

        if bind is None:
            bind = cls.__metadata__.bind
        row = await bind.first(clause)

        await self.after_update(cls, bind, self, row)

        if not row:
            raise NoSuchRowError()
        for k, v in row.items():
            self._instance.__values__[
                self._instance._column_name_map.invert_get(k)] = v
        for prop in self._props:
            prop.reload(self._instance)
        return self

...

_delete ...

        if bind is None:
            bind = self.__metadata__.bind
        data = (await bind.status(clause))[0]
        await self.after_delete(cls, bind, self, data)
        return data

...

app_register_events.py

from sqlalchemy import event
from myapp.models import UserModel

async def after_create_method(model_class, db_bind, instance, row, **kw):
    .... something async
    return audit_operation

event.listen(UserModel.__table__, "after_create", after_create_method)

[wwwjfy]

Thanks for this!

I haven't used SQLAlchemy's event system much, so allow me take a closer look.

One concern from my first skim is I'm not sure if we should await for the registered functions to be executed. Because, strictly speaking, a event should not block or fail the main process.

Refs #161

[tng10]

Yeah, you're absolutely right, it shouldn't block the main process. How about creating a task by that moment?

import asyncio

...
class GinoSQLAlchemyCoreEvents:

    async def call_method(self, method_to_call, *args, **kwargs):
            if asyncio.iscoroutinefunction(method_to_call):
                loop = asyncio.get_event_loop()
                loop.create_task(method_to_call(*args, **kwargs))
            else:
                method_to_call(*args, **kwargs)

[fantix]

Ah, I see, so it is about adding new events to CRUD, not dealing with the existing SQLAlchemy engine/connection/execution events. I'm with you to add this feature, but let's put it in post 1.0 releases, and your cool mixin would work before that. Thanks for the idea and code!

For the blocking/forking topic, I think we should somehow align with SQLAlchemy - callbacks are executed in the same transaction/session for connection events or session events. Perhaps it should be an option for the user to create a task, not a default. Thoughts pls?

[tng10]

Perfect, I think by having this functionality on CRUD class would be pretty good. There's also the possibility to write against to the database using other methods rather than CRUD's methods, for instance when executing something like

UserModel.update.values(
        nickname='No.' + db.cast(User.id, db.Unicode),
    ).where(
        User.id > 10,
    ).gino.status()

Events could be also attached to this method and by this way covering bulk updates.

The connection events provided by SQLAlchemy Core are quite interesting and aligning some of these events it with Gino sounds very good to me, regarding the session events I don't know what is the equivalent on Gino's side, we don't a have a session, do we? 🤔

Offering the user to create a task is more explicit and might be the better approach, since blocking/forking topic can vary according to each scenario.

[fantix]

regarding the session events I don't know what is the equivalent on Gino's side, we don't a have a session, do we? 🤔

No, we don't have the concept of Session or equivalent, sorry I should have referenced to the mapper events, like after_insert, after_update and after_delete.

[tng10]

Ahh alright! :)