SQLAlchemy：如何确保泛型关联中Address必关联父类？

我参考SQLAlchemy的泛型关联示例，采用了table per association的实现方式，通过Mixin为每个父类生成独立关联表，所有地址数据存储在单个Address表中。现在需要保证Address的每一行必须关联到Customer或Supplier，类似抽象类的约束效果，请问如何实现？

原示例代码：

from sqlalchemy import Column, Integer, String, ForeignKey, Table, create_engine
from sqlalchemy.ext.declarative import as_declarative, declared_attr
from sqlalchemy.orm import relationship, Session

@as_declarative()
class Base(object):
    """Base class which provides automated table name
    and surrogate primary key column.
    """

    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()

    id = Column(Integer, primary_key=True)


class Address(Base):
    """The Address class.
    This represents all address records in a
    single table.
    """

    street = Column(String)
    city = Column(String)
    zip = Column(String)

    def __repr__(self):
        return "%s(street=%r, city=%r, zip=%r)" % (
            self.__class__.__name__,
            self.street,
            self.city,
            self.zip,
        )


class HasAddresses(object):
    """HasAddresses mixin, creates a new address_association
    table for each parent.
    """

    @declared_attr
    def addresses(cls):
        address_association = Table(
            "%s_addresses" % cls.__tablename__,
            cls.metadata,
            Column("address_id", ForeignKey("address.id"), primary_key=True),
            Column(
                "%s_id" % cls.__tablename__,
                ForeignKey("%s.id" % cls.__tablename__),
                primary_key=True,
            ),
        )
        return relationship(Address, secondary=address_association)


class Customer(HasAddresses, Base):
    name = Column(String)


class Supplier(HasAddresses, Base):
    company_name = Column(String)


engine = create_engine("sqlite://", echo=True)
Base.metadata.create_all(engine)

session = Session(engine)

session.add_all(
    [
        Customer(
            name="customer 1",
            addresses=[
                Address(
                    street="123 anywhere street", city="New York", zip="10110"
                ),
                Address(
                    street="40 main street", city="San Francisco", zip="95732"
                ),
            ],
        ),
        Supplier(
            company_name="Ace Hammers",
            addresses=[
                Address(street="2569 west elm", city="Detroit", zip="56785")
            ],
        ),
    ]
)

session.commit()

for customer in session.query(Customer):
    for address in customer.addresses:
        print(address)

要实现Address必须关联至少一个父类的约束，需要从数据库层面和ORM层面双重限制：

1. 数据库级CHECK约束（强制数据完整性） #

通过表级CHECK约束，确保Address的id至少存在于customer_addresses或supplier_addresses关联表中，即使绕过ORM直接操作数据库也无法创建孤行地址。

2. ORM层面限制直接实例化Address #

禁用Address的构造函数，强制只能通过父类（Customer/Supplier）的addresses关系创建地址实例，从源头避免无关联地址的生成。

修改后的完整代码 #

from sqlalchemy import (
    Column, Integer, String, ForeignKey, Table, CheckConstraint,
    create_engine
)
from sqlalchemy.ext.declarative import as_declarative, declared_attr
from sqlalchemy.orm import relationship, Session


@as_declarative()
class Base(object):
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()

    id = Column(Integer, primary_key=True)


class Address(Base):
    street = Column(String)
    city = Column(String)
    zip = Column(String)

    # 添加表级CHECK约束，确保地址至少关联一个父类
    __table_args__ = (
        CheckConstraint(
            "EXISTS(SELECT 1 FROM customer_addresses WHERE address_id = id) "
            "OR EXISTS(SELECT 1 FROM supplier_addresses WHERE address_id = id)",
            name="address_must_have_parent"
        ),
    )

    def __repr__(self):
        return "%s(street=%r, city=%r, zip=%r)" % (
            self.__class__.__name__,
            self.street,
            self.city,
            self.zip,
        )

    # 禁止直接实例化Address
    def __init__(self, *args, **kwargs):
        raise NotImplementedError(
            "Address不能直接实例化，请通过Customer.addresses或Supplier.addresses创建"
        )


class HasAddresses(object):
    @declared_attr
    def addresses(cls):
        address_association = Table(
            "%s_addresses" % cls.__tablename__,
            cls.metadata,
            Column("address_id", ForeignKey("address.id"), primary_key=True),
            Column(
                "%s_id" % cls.__tablename__,
                ForeignKey("%s.id" % cls.__tablename__),
                primary_key=True,
            ),
        )
        return relationship(
            Address,
            secondary=address_association,
            uselist=True
        )


class Customer(HasAddresses, Base):
    name = Column(String)

    # 自定义地址构造逻辑，绕开Address的__init__限制
    @declared_attr
    def addresses(cls):
        rel = super().addresses
        def init_address(**kwargs):
            addr = Address.__new__(Address)
            for k, v in kwargs.items():
                setattr(addr, k, v)
            return addr
        rel.constructor = init_address
        return rel


class Supplier(HasAddresses, Base):
    company_name = Column(String)

    @declared_attr
    def addresses(cls):
        rel = super().addresses
        def init_address(**kwargs):
            addr = Address.__new__(Address)
            for k, v in kwargs.items():
                setattr(addr, k, v)
            return addr
        rel.constructor = init_address
        return rel


# SQLite需开启check_constraints参数以启用CHECK约束
engine = create_engine(
    "sqlite://",
    echo=True,
    connect_args={"check_same_thread": False, "check_constraints": True}
)
Base.metadata.create_all(engine)

session = Session(engine)

# 测试：直接创建Address会抛出错误
try:
    addr = Address(street="Test", city="Test", zip="123")
    session.add(addr)
    session.commit()
except NotImplementedError as e:
    print(f"拦截直接创建Address: {e}")
    session.rollback()

# 正确创建方式：通过父类的addresses属性
session.add_all(
    [
        Customer(
            name="customer 1",
            addresses=[
                {"street": "123 anywhere street", "city": "New York", "zip": "10110"},
                {"street": "40 main street", "city": "San Francisco", "zip": "95732"},
            ],
        ),
        Supplier(
            company_name="Ace Hammers",
            addresses=[
                {"street": "2569 west elm", "city": "Detroit", "zip": "56785"}
            ],
        ),
    ]
)
session.commit()

# 测试：删除所有关联会触发CHECK约束错误
try:
    customer = session.query(Customer).first()
    customer.addresses = []
    session.commit()
except Exception as e:
    print(f"拦截孤行Address: {e}")
    session.rollback()

# 验证查询结果
for customer in session.query(Customer):
    for address in customer.addresses:
        print(address)

for supplier in session.query(Supplier):
    for address in supplier.addresses:
        print(address)

关键说明 #

1. 数据库兼容性：MySQL 5.7及以下版本不支持CHECK约束（会被忽略），需用触发器替代；PostgreSQL、SQLite（开启check_constraints）、MySQL 8.0+完全支持。
2. 级联删除（可选）：如果需要删除父类时自动删除关联地址，可在relationship中添加cascade="all, delete-orphan"参数。
3. 约束优先级：数据库级约束是最后一道防线，ORM层面限制则是开发阶段的前置拦截，两者结合确保数据完整性。

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