Question 263669 in PostgreSQL Question posted in
The official documentation can be found here.

Postgresql – sqlalchemy listens_for() doesn't trigger

I have working listens_for function, which inserts LoadHistory object before updating Load:

@event.listens_for(Load, 'before_update')
def before_update_load(mapper, connection, target):
    current_load_select = select(load_fields).where(Load.id == target.id)
    insert_history = insert(LoadHistory).from_select(load_history_fields, current_load_select)
    connection.execute(insert_history)

And the task is to have similar function but which will insert LoadStopHistory objects after LoadHistory object has been inserted. The code is:

@event.listens_for(LoadHistory, 'after_insert')
def after_insert_load_history(mapper, connection, target):
    # here will be code
    print("@" * 125)
    print('trgt', target)
    print("@" * 125)

but the issue that this function has never been called.
I’ve tried to insert LoadHistory with updating Load or directly with Postgres, neither practice had any result.
I’ve tried with ‘after_insert’ changed to ‘before_insert’ and ‘before_update’.

Tags:

2

Answers


  1. 0

    There are various possible causes to look into if your event listener for the after_insert event on the LoadHistory model is not being called. The relationships between your SQLAlchemy models (Load, LoadHistory, and LoadStopHistory) should be properly specified first because events may not occur in the absence of such relationships. Second, keep in mind that events are frequently started during the SQLAlchemy session commit, so be sure to do so after entering a "LoadHistory" object. Make that your event listeners are registered correctly and are executed when your application launches. Be mindful of database transactions; if one is reversed before the session commits, it may prevent events from happening.

    Hope this helps, do let me know if you want help

    Login or Signup to reply.
  2. 0

    The impression I get from this note (from Mapper-level Flush Events):

    It is important to note that these events apply only to the session flush operation , and not to the ORM-level INSERT/UPDATE/DELETE functionality described at ORM-Enabled INSERT, UPDATE, and DELETE statements. To intercept ORM-level DML, use the SessionEvents.do_orm_execute() event.

    is that the after_insert event will not register the result of calling insert() directly – the do_orm_execute listener must be used instead. Moreover, the insert should be executed using the session rather than the connection to be registered by the ORM event machinery.

    These listeners seem to work:

    import sqlalchemy as sa

...

@sa.event.listens_for(Load, 'before_update')
def before_update_load(mapper, connection, target):
    insert_history = sa.insert(LoadHistory)
    session = sa.inspect(target).session
    session.execute(insert_history)

@sa.event.listens_for(Session, 'do_orm_execute')
def receive_orm_execute(orm_execute_state):
    if (
        orm_execute_state.is_insert
        and orm_execute_state.statement.table.name
        == LoadHistory.__table__.name
    ):
        print('*' * 10)
    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search