SqlAlchemy 'before_insert' event is not triggered with async setup

[mcostacurta-fulltruck]

As per the the title, I have this setup for an Order class:

class Order(Base):
    __tablename__ = 'order'

    id = Column(pgUUID(as_uuid=True), primary_key=True, default=uuid4, unique=True, nullable=False)
    user_id = Column(String, ForeignKey('user.id'), nullable=False)

    number = Column(Integer, default=0) 

    status = Column(Enum(Status), nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

I'm trying to intercept the 'before_insert' event with this code:

@event.listens_for(Order, 'before_insert')
async def before_insert(mapper, connection, target):
    # Use the session linked with the target, ensure it's synchronous
    async_session = AsyncSession(connection)
    
    # Extract the current year from created_at or use the current year if not set
    curr_year = target.created_at.year if target.created_at else datetime.now().year
    
    # Async query to find the current maximum number for the year
    result = await async_session.execute(select(func.max(Order.number)).filter(func.extract('year', Order.created_at) == curr_year))
    max_number = result.scalar_one_or_none()
    
    # Set the number for the current order
    target.number = (max_number or 0) + 1
    
    # Close the async session
    await async_session.close()


engine = create_async_engine(DATABASE_URL, future=True, echo=False, pool_size=30, max_overflow=50)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

async with async_session() as session:
        async with session.begin():
            new_order = Order(**kwargs) # kwargs are some info to create the orders
            session.add(new_order)
            await session.flush()

Nevertheless, the event is never triggered. Is there anything I'm doing wrong? Can Sqlalchemy events be used in an asynchronous environment?

[zzzeek]

you can use the events but they operate using Python code that's written in synchronous style (nevertheless, it is still run within a context that is "async" and is using your async driver directly, you just dont see it). so your def would not be async, and you would use non-async Session inside your event hook to do work.

docs for this use case are at https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html#using-events-with-the-asyncio-extension

[gordthompson]

As for the event never being triggered, that appears to be because …

site-packages/sqlalchemy/orm/events.py:887: RuntimeWarning: coroutine 'async_main.<locals>.before_insert' was never awaited
  fn(*arg, **kw)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

… i.e., your async def was not awaited. Just using def for the listener does get us into the routine (verified by adding a breakpoint() line in the function and getting dropped into pdb).

[mcostacurta-fulltruck]

Got it and makes sense, thank you!

[Fidel-C]

Have you solved this. If yes, how please?