SQLAlchemy能否将多对多关联表映射为字符串键列表？

我有两个存在多对多关系的数据库表，表定义如下：

purchase_order = Table(
    "purchase_order",
    metadata,
    Column("name", String, nullable=False, primary_key=True),
    ...
)

shipment = Table(
    "shipment",
    metadata,
    Column("identifier", String, nullable=False, primary_key=True),
)

shipment_po = Table(
    "shipment_po",
    metadata,
    Column(
        "shipment_identifier",
        String,
        ForeignKey("shipment.identifier"),
        nullable=False,
        primary_key=True,
    ),
    Column(
        "purchase_order_name",
        String,
        ForeignKey("purchase_order.name"),
        nullable=False,
        primary_key=True,
    ),
)

我希望将这些表映射为如下Shipment类（注意purchase_order_names是字符串列表，而非PurchaseOrder对象）：

class Shipment:
    def __init__(
        self,
        identifier: str,
        purchase_order_names: list[str],
    ):
        ...

以便实现以下操作：

• 创建时：

  session.add(Shipment("my_shipment", purchase_order_names=["my_po_1", "my_po_2"]))
  

• 查询时：

  session.query(Shipment).filter_by(identifier="my_shipment").one()
  

• 更新时：

  shipment.purchase_order_names.append("my_po_3")
  

请问SQLAlchemy是否支持这种映射方式？

SQLAlchemy完全支持这种映射方式，你可以用**关联代理（Association Proxy）**配合基础的关系映射来实现，或者自定义属性的getter/setter方法封装底层多对多操作。

实现方案示例（声明式映射） #

假设使用SQLAlchemy的声明式映射模式，具体实现如下：

from sqlalchemy import Column, String, ForeignKey, Table
from sqlalchemy.orm import relationship, declarative_base, association_proxy

Base = declarative_base()

# 复用你定义的三张表结构
purchase_order = Table(
    "purchase_order",
    Base.metadata,
    Column("name", String, nullable=False, primary_key=True),
    # 其他字段...
)

shipment_po = Table(
    "shipment_po",
    Base.metadata,
    Column(
        "shipment_identifier",
        String,
        ForeignKey("shipment.identifier"),
        nullable=False,
        primary_key=True,
    ),
    Column(
        "purchase_order_name",
        String,
        ForeignKey("purchase_order.name"),
        nullable=False,
        primary_key=True,
    ),
)

# 定义PurchaseOrder模型（即使不直接操作，也需要它关联中间表）
class PurchaseOrder(Base):
    __table__ = purchase_order

class Shipment(Base):
    __table__ = Table(
        "shipment",
        Base.metadata,
        Column("identifier", String, nullable=False, primary_key=True),
    )

    # 底层多对多关系，关联到PurchaseOrder对象（对外隐藏）
    _purchase_orders = relationship(
        "PurchaseOrder",
        secondary=shipment_po,
        backref="shipments"
    )

    # 关联代理：自动转换PurchaseOrder对象与name字符串列表
    purchase_order_names = association_proxy(
        "_purchase_orders",
        "name",
        creator=lambda name: PurchaseOrder(name=name)
    )

    def __init__(self, identifier: str, purchase_order_names: list[str]):
        self.identifier = identifier
        self.purchase_order_names = purchase_order_names

验证你需要的操作 #

• 创建：

  # 假设已初始化好session
  session.add(Shipment("my_shipment", purchase_order_names=["my_po_1", "my_po_2"]))
  session.commit()
  

• 查询：

  shipment = session.query(Shipment).filter_by(identifier="my_shipment").one()
  print(shipment.purchase_order_names)  # 输出: ["my_po_1", "my_po_2"]
  

• 更新：

  shipment.purchase_order_names.append("my_po_3")
  session.commit()
  # 中间表shipment_po会自动新增关联记录
  

关键说明 #

• _purchase_orders是底层的关系属性，负责处理与PurchaseOrder的多对多关联，对外隐藏避免直接操作对象。
• purchase_order_names通过association_proxy实现自动转换：读取时提取每个PurchaseOrder的name组成列表；写入时（初始化、append操作）自动创建对应的PurchaseOrder对象（如果需要确保采购订单已存在，可以修改creator逻辑或提前做存在性校验）。

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