python - 将 contains_eager 与过滤器一起使用，但不过滤掉没有匹配子代的父代

Question

这个问题已经有答案了:

How to eager load optionally exists filtered relation (1 个回答)

已关闭 4 年前。

阅读 SQLAlchemy 文档:

http://docs.sqlalchemy.org/en/latest/orm/loading_relationships.html#using-contains-eager-to-load-a-custom-filtered-collection-result

The above query will load only User objects which contain at least Address object that contains the substring 'ed' in its email field.

我正在尝试做类似的事情，但我也想检索所有根本没有匹配地址的 User 对象。

为了更好地解释自己，我修改了这段代码，这是 SQLAlchemy 文档中的示例之一:

from sqlalchemy.ext.declarative import as_declarative, declared_attr
from sqlalchemy import create_engine, Integer, Column, \
                    String, and_ 
from sqlalchemy.orm import Session, relationship, foreign, remote, backref, joinedload, contains_eager
from sqlalchemy import event


@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)

    discriminator = Column(String)
    """Refers to the type of parent."""

    parent_id = Column(Integer)
    """Refers to the primary key of the parent.

    This could refer to any table.
    """

    @property
    def parent(self):
        """Provides in-Python access to the "parent" by choosing
        the appropriate relationship.

        """
        return getattr(self, "parent_%s" % self.discriminator)

    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 relationship to
    the address_association table for each parent.

    """

@event.listens_for(HasAddresses, "mapper_configured", propagate=True)
def setup_listener(mapper, class_):
    name = class_.__name__
    discriminator = name.lower()
    class_.addresses = relationship(Address,
                        primaryjoin=and_(
                                        class_.id == foreign(remote(Address.parent_id)),
                                        Address.discriminator == discriminator
                                    ),
                        backref=backref(
                                "parent_%s" % discriminator,
                                primaryjoin=remote(class_.id) == foreign(Address.parent_id)
                                )
                        )
    @event.listens_for(class_.addresses, "append")
    def append_address(target, value, initiator):
        value.discriminator = discriminator

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")
        ]
    ),
    Customer(
        name='customer 2',
        addresses=[
            Address(
                    street='88 hello yes',
                    city="Old Potato",
                    zip="10110"),
            Address(
                    street='123 banana street',
                    city="New Texas",
                    zip="23422"),
            Address(
                    street='40 dsfasadfmain street',
                    city="San Cocisfran",
                    zip="10110")
        ]
    ),
    Customer(
        name='customer 3',
        addresses=[
            Address(
                    street='123 orange street',
                    city="New Mew",
                    zip="23422"),
        ]
    ),
    Supplier(
        company_name="Ace Hammers",
        addresses=[
            Address(
                    street='2569 west elm',
                    city="Detroit",
                    zip="56785")
        ]
    ),
])

session.commit()

query = (
    session.query(Customer)
    .outerjoin(Customer.addresses)
    .filter(Address.zip == "10110")
    .options(contains_eager(Customer.addresses))
)
for customer in query.all():
    print(customer.name)
    print(str(customer.addresses))

此代码正确打印了客户 1 和 2 及其邮政编码为 10110 的地址，但我也想获取客户 3，即使它根本没有匹配的地址。

即:我想要获取所有客户的列表，无论他们的地址是什么，对于每个客户，我希望填充他们的地址，但前提是他们的邮政编码为 10110。

理想情况下，这将显示为 JOIN 的附加条件，而不是 WHERE。结果查询可能看起来像这样，我只是不知道如何用 SQLAlchemy 术语“表达”它:

... LEFT OUTER JOIN address ON customer.id = address.parent_id AND address.discriminator = 'customer' AND address.zip = '10110'

Answer 1

您可以更改连接条件:

query = session.query(Customer) \
               .outerjoin(Address, and_(Address.parent_id == Customer.id,
                                        Address.discriminator == "customer",
                                        Address.zip == "10110")) \
               .options(contains_eager(Customer.addresses))