SqlAlchemy异步环境中before_insert事件未触发问题排查

一、事件未触发的核心原因 #

你的代码存在两个关键错误，导致异步环境下before_insert事件无法正常执行：

1. 未声明异步事件标识 #

SQLAlchemy对同步、异步事件回调做了区分，当你用async def定义事件处理函数时，必须在@event.listens_for装饰器中添加asyncio=True参数，否则框架会将其当作同步函数处理，无法在异步上下文里触发执行。

2. 错误创建独立AsyncSession #

事件回调中已经传入了当前事务的connection对象，无需再新建AsyncSession——新建的session会脱离当前事务上下文，既可能查询不到同批次未提交的数据，还会干扰事件的正常执行逻辑。应直接使用传入的connection执行异步查询。

二、修正后的代码实现 #

from sqlalchemy import func, extract
from sqlalchemy.ext.asyncio import AsyncConnection
from sqlalchemy.event import listens_for
import datetime
from datetime import timezone

@listens_for(Order, 'before_insert', asyncio=True)  # 必须添加asyncio=True标识
async def before_insert(mapper, connection: AsyncConnection, target):
    # 获取当前年份：优先使用target的created_at，否则取数据库服务器的当前年份（避免客户端时区偏差）
    if target.created_at:
        curr_year = target.created_at.year
    else:
        # 通过数据库函数获取年份，保证与created_at的时区一致
        year_result = await connection.execute(select(func.extract('year', func.now())))
        curr_year = int(year_result.scalar_one())
    
    # 直接使用当前connection执行查询，无需新建session
    result = await connection.execute(
        select(func.max(Order.number))
        .filter(extract('year', Order.created_at) == curr_year)
    )
    max_number = result.scalar_one_or_none()
    
    target.number = (max_number or 0) + 1

# 引擎与Session初始化代码保持不变
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)
        session.add(new_order)
        await session.flush()

三、异步环境下SQLAlchemy事件的使用规则 #

SQLAlchemy完全支持异步场景的事件监听，需遵守以下规则：

• 异步事件回调必须用async def定义，且在@event.listens_for中指定asyncio=True。
• 事件回调中执行数据库操作时，必须使用传入的AsyncConnection对象，不能使用同步Connection，也不要随意新建Session。
• 所有IO操作（如查询、执行语句）必须通过await调用。

补充：并发风险提示 #

你当前的订单编号生成逻辑存在并发冲突风险：当多个请求同时创建订单时，可能会同时查询到相同的max_number，导致生成重复编号。常见的解决思路：

• 单独创建年份计数器表，使用FOR UPDATE锁或数据库原子操作保证编号唯一性。
• 利用PostgreSQL的序列（Sequence）结合年份前缀实现自增逻辑。

内容的提问来源于stack exchange，提问作者Gad82