如何在SQLAlchemy中用back_populates处理双外键关联？

错误原因 #

Inventory与Transfer之间存在两个外键关联（来源库存ID、目的地库存ID），SQLAlchemy无法自动推断Inventory.transfers_origin和Inventory.transfers_destination分别对应哪个外键，因此抛出AmbiguousForeignKeysError。

修改后的模型代码 #

from sqlalchemy import ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship, DeclarativeBase
import datetime

class Base(DeclarativeBase):
    pass

class Inventory(Base):
    __tablename__ = "inventories"
    inventory_id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(nullable=False)
    # 明确指定关联Transfer的inventory_origin_id外键
    transfers_origin: Mapped[List["Transfer"]] = relationship(
        back_populates="inventory_origin",
        foreign_keys="Transfer.inventory_origin_id"
    )
    # 明确指定关联Transfer的inventory_destination_id外键
    transfers_destination: Mapped[List["Transfer"]] = relationship(
        back_populates="inventory_destination",
        foreign_keys="Transfer.inventory_destination_id"
    )

class Transfer(Base):
    __tablename__ = "transfers"
    transfer_id: Mapped[int] = mapped_column(primary_key=True)
    date: Mapped[datetime.datetime] = mapped_column(server_default=func.CURRENT_TIMESTAMP())
    inventory_origin_id: Mapped[int] = mapped_column(ForeignKey("inventories.inventory_id"))
    inventory_origin: Mapped["Inventory"] = relationship(
        foreign_keys=[inventory_origin_id],
        back_populates="transfers_origin"
    )
    inventory_destination_id: Mapped[int] = mapped_column(ForeignKey("inventories.inventory_id"))
    inventory_destination: Mapped["Inventory"] = relationship(
        foreign_keys=[inventory_destination_id],
        back_populates="transfers_destination"
    )

关键修改说明 #

• 在Inventory.transfers_origin中添加foreign_keys="Transfer.inventory_origin_id"，明确该关系对应转移记录的来源库存ID外键
• 在Inventory.transfers_destination中添加foreign_keys="Transfer.inventory_destination_id"，明确该关系对应转移记录的目的地库存ID外键

异步查询示例 #

使用AsyncSession查询指定库存的来源/目的地转移记录：

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.future import select

# 初始化异步引擎与会话
async_engine = create_async_engine("sqlite+aiosqlite:///inventory.db")

# 查询指定库存作为来源的所有转移记录
async def get_origin_transfers(inventory_id: int):
    async with AsyncSession(async_engine) as session:
        stmt = select(Transfer).where(Transfer.inventory_origin_id == inventory_id)
        result = await session.execute(stmt)
        return result.scalars().all()

# 查询指定库存作为目的地的所有转移记录
async def get_destination_transfers(inventory_id: int):
    async with AsyncSession(async_engine) as session:
        stmt = select(Transfer).where(Transfer.inventory_destination_id == inventory_id)
        result = await session.execute(stmt)
        return result.scalars().all()

内容的提问来源于stack exchange，提问作者Luis Rodolfo Molina Molina