Is it possible to get the entity from SQLAlchemys "do_orm_execute" event?

Question

We're trying to implement a History table for some select classes. All the queries are written for SQLAlchemy Core 1.x and passed to the execute() method on the Session object. That means we cannot use the more straight forward events such as "before_flush".

According to the docs we should be able to access the type that is a target for an insert(), update() or delete()

But that just throws AttributeError: 'Update' object has no attribute 'column_descriptions' (I'm guessing it's either due to it only being available on selects or only if the query originated from the ORM)

All the tables are declared in a file generated from sqlacodegen without any declarative base, like so:

t_example = Table(
    'examples', metadata,
    Column('id', UUID, primary_key=True),
    Column('name', Text, nullable=False),
)

Now we've newly migrated to SQLAlchemy 2.0 and implemented asyncio, implemented a declarative_base but since thousands of queries are written directly towards tables like:

query = select(t_example.c.name).where(
        t_example.c.id == some_id
    )

And we don't want to spend weeks updating them but slowly migrating the tables to classes like so:

class Example(Base, LogInsertMixin, LogUpdateMixin, LogDeleteMixin):
    __table__ = t_example
    id: Mapped[UUID]
    name: Mapped[str]

As you can see we have some mixins that I thought could be declared on the models that require some logging to the database:

class LogInsertMixin:
    pass


class LogUpdateMixin:
    pass


class LogDeleteMixin:
    pass

But I can't find any way in the "do_orm_execute" event to make an isistance() check toward:

class DatabaseLogger:

    def orm_execute_event(self, orm_execute_state: ORMExecuteState):
        if orm_execute_state.is_update and isinstance(*****what to check for? :(*****, LogUpdateMixin):
            pass

I can see that the event fires but ORMExecuteState isn't well documented. Is this way even feasible? How can I find out what Table() was passed to an update() or delete()?

Answer 1

To anyone falling into this rabbit hole I found the solution.

orm_execute_state has a property statement which is an Executable, in my case it will be either a construct of Select, Delete, Update or Insert.

On those construts we can access entity_description which contains the mapping:

class DatabaseLogger:

def orm_execute_event(self, orm_execute_state: ORMExecuteState):
    if orm_execute_state.is_update and isinstance(
        orm_execute_state.statement, Update
    ): # isinstance to tell vscode to give correct code hints
        print(
            orm_execute_state.statement.entity_description
        )

Results in:

{'name': 'Example', 'type': <class 'Example'>, 'expr': <class 'Example'>, 'entity': <class 'Example'>, 'table': Table('t_example', MetaData(), Column('id', UUID(), table=<t_example>, primary_key=True, nullable=False, server_default=DefaultClause(<sqlalchemy.sql.elements.TextClause object at 0x7fbcd020d7c0>, for_update=False)), Column('name', Text(), table=<t_example>))}