SQLAlchemy 1.4异步事件侦听器

Question

尝试使用事件侦听器(https://docs.sqlalchemy.org/en/14/core/events.html#sqlalchemy.events.ConnectionEvents)在异步sqlalchemy引擎上，我得到了以下错误：{NotImplementedError}asynchronous events are not implemented at this time. Apply synchronous listeners to the AsyncEngine.sync_engine or AsyncConnection.sync_connection attributes.

如果我正确理解这一点，我就不能在异步引擎上使用事件，如果我想要事件支持，我必须切换到同步引擎？

engine: Engine = create_async_engine(
URL, echo=True, future=True
)
async_session = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False, future=True
)

event.listens_for(engine, "do_connect")(do_connect_listener)
event.listens_for(engine, "engine_connect")(engine_connect_listener)

Answer 1

从github上的SQLAlchemy讨论部分得到了答案。OP的所有信用(https://github.com/sqlalchemy/sqlalchemy/discussions/6594#discussioncomment-836437(：

"你好，

根据异常的建议，在应用事件时可以使用sync_engine/sync_connection。这里有一个例子：；

import asyncio
from sqlalchemy import event
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
@event.listens_for(engine.sync_engine, "do_connect")
def do_connect(dialect, conn_rec, cargs, cparams):
print("some-function")
@event.listens_for(engine.sync_engine, "engine_connect")
def engine_connect(conn, branch):
print("engine_connect", conn.exec_driver_sql("select 1").scalar())
async def go():
async with engine.connect() as conn:
res = await conn.exec_driver_sql("select 2")
print("go", res.scalar())
asyncio.run(go())