在 SQLAlchemy 中为映射表创建事件监听器

Question

我的应用程序中有两个模型。

class Building(BaseModel):
    __tablename__ = "buildings"

    address_id = mapped_column(BigInteger, ForeignKey("addresses.id"), nullable=False)
    name = mapped_column(String(6), nullable=True, default=None)

    address = relationship("Address", foreign_keys=[address_id], lazy="raise")
    contacts = relationship("Contact", secondary=BuildingContacts, lazy="raise")

和

class Contact(BaseModel):
    __tablename__ = "contacts"

    type = mapped_column(Enum(ContactTypes, name="contact_types"), nullable=False)
    value = mapped_column(String(50), nullable=False)

我的 BaseModel 看起来像：

engine: AsyncEngine = create_async_engine(
    app_settings.db.database_url,
    echo=app_settings.debug,
    echo_pool=app_settings.debug,
    pool_size=app_settings.db.pool_size,
    max_overflow=0,
    pool_pre_ping=True,
)
async_session: async_sessionmaker[AsyncSession] = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)
database = databases.Database(app_settings.db.database_url)


async def get_session() -> AsyncGenerator[AsyncSession, AsyncSession]:
    async with async_session() as session:
        yield session


class Base(DeclarativeBase):
    pass


class BaseModel(Base):
    __abstract__ = True

    id = mapped_column(sa.BigInteger, primary_key=True)
    uuid = mapped_column(UUID(as_uuid=False), default=lambda: str(uuid4()), unique=True, nullable=False)

    created_at = mapped_column(sa.DateTime(timezone=True), server_default=sa.sql.func.now())
    updated_at = mapped_column(sa.DateTime(timezone=True), server_default=sa.sql.func.now(), onupdate=sa.sql.func.now())

没有什么特别的，但我也有地图表来连接建筑物和联系人（多2多），它看起来像：

BuildingContacts = Table(
    "building_contacts",
    Base.metadata,
    Column("building_id", BigInteger, ForeignKey("buildings.id"), nullable=False),
    Column("contact_id", BigInteger, ForeignKey("contacts.id"), nullable=False),
)

我的问题是：我无法以经典方式为 BuildingContacts 创建事件侦听器：

@event.listens_for(BuildingContacts, "after_insert")
def after_insert_listener(mapper, connection, target):
    print(f"Inserted new record in BuildingContacts: {target}")

它无法与任何其他带有错误“AttributeError：after_insert”的标识符一起使用。 有人可以解释一下，我怎样才能听这些地图表吗？

我尝试使用任何其他标识符，但我总是捕获 AttributeError

Answer 1

看来此事件可能不会针对表触发，而仅针对映射类触发。

• 也许您可以尝试观看一个事件，例如

  "after_update"

  上的

  Contact

  ，并检查关系是否发生变化。 它似乎跟踪关系的变化，但我不确定从该事件中可以获得多少信息。
• 或者将辅助表提升为完整的 Orm 类并跟踪其上的事件。