动态滤除sqlalchemy中的数据

Question

我有这些型号

class Integration(Base, ModelBase, ModelSerializer):
    __tablename__ = 'integration'

    id = Column(Integer, primary_key=True)
    domain = relationship('Domain', backref='integration')
    created_at = Column(DateTime, default=datetime.now())
    updated_at = Column(DateTime, default=datetime.now())
    name = Column(String(512), index=True)
    meta = Column(JSON)
    tag = Column(String(512))
    identifier_id = Column(String(512), unique=True, index=True)

    def __repr__(self):
        return '<name {}>'.format(self.name)


class Domain(Base, ModelBase, ModelSerializer):
    __tablename__ = 'domain'

    id = Column(Integer, primary_key=True)
    integration_id = Column(Integer, ForeignKey('integration.id'), index=True)
    url = relationship('Url', backref='domain')
    created_at = Column(DateTime, default=datetime.now())
    updated_at = Column(DateTime, default=datetime.now())
    name = Column(String(512), index=True)
    domain = Column(String(512), nullable=True)

    def __repr__(self):
        return '<name {}>'.format(self.name)


class Url(Base, ModelBase, ModelSerializer):
    __tablename__ = 'url'

    id = Column(Integer, primary_key=True)
    domain_id = Column(Integer, ForeignKey('domain.id'), index=True)
    created_at = Column(DateTime, default=datetime.now())
    updated_at = Column(DateTime, default=datetime.now())
    name = Column(String(512), index=True)
    url = Column(String(512), nullable=True)
    meta = Column(JSON)
    auth = Column(JSON)

    def __repr__(self):
        return '<name {}>'.format(self.name)

class ModelBase(object):

    def __repr__(self):
        return '<id {}>'.format(self.id)

class ModelSerializer(object):

    def as_dict(self):
        return {attribute.name: getattr(self, attribute.name) for attribute in self.__table__.columns}

现在我需要根据过滤器过滤出数据。如果假设给出了integration_filters，则仅过滤Integration数据。如果给出了Integration和Domain，则通过加入它们仅过滤域和集成与网址和域相同

我尝试过的解决方案-

1）。不工作，因为它不联接然后表。后来发现add_entity不会添加模型。

        if integration_filter:
            query = query.add_entity(Integration)
            query = query.filter_by(**kwargs['integration_filter'])
        if domain_filter:
            query = query.add_entity(Domain)
            query = query.filter_by(**kwargs['domain_filter'])
        if url_filter:
            query = query.add_entity(Url)
            query = query.filter_by(**kwargs['url_filter'])

2）。它在集成表中检查属性“域”]

    if integration_filter:
        query = session.query(Integration).filter_by(**integration_filter)

    if domain_filter:
        if query:
            query = query.join(Domain)
        else:
            query = session.query(Domain)

        query = query.filter_by(**domain_filter)

    if url_filter:

        if query:
            query = query.join(Url)
        else:
            query = session.query(Url)

        query = query.filter_by(**url_filter)`
3）。这也不起作用]

models = []
joins = []

if integration_filter:
    models.append(Integration)

if domain_filter:
    if models:
        joins.append((Integration, Domain, Integration.id == Domain.integration_id))
    models.append(Domain)

if url_filter:
    if models:
        joins.append((Domain, Url, Domain.id == Url.domain_id))
    models.append(Url)

query = session.query(*models)

for join in joins:
    query = query.join(*join)

我有这些模型类Integration（Base，ModelBase，ModelSerializer）：__tablename__ ='integration'id = Column（Integer，primary_key = True）domain = Relationship（'Domain'，backref ='...

Answer 1

[这个工作。虽然很简单，但一开始没想到

try:

    session = Backend().get_session()

    models = []
    join_filter = []

    if integration_filter:
        models.append(Integration)
        join_filter.append({'filter': integration_filter})

    if domain_filter:
        models.append(Domain)
        join_filter.append({'join': (Domain, Domain.integration_id == Integration.id), 'filter': domain_filter})

    if url_filter:
        models.append(Url)
        join_filter.append({'join': (Url, Url.domain_id == Domain.id), 'filter': url_filter})

    query = session.query(*models)

    for jf in join_filter:
        if 'join' in jf:
            query = query.join(jf['join'])
        if 'filter' in jf:
            query = query.filter_by(**jf['filter'])

    query = query.__getattribute__(query_type)

    self.records = query()

except InvalidRequestError as ir:
    raise Exception("Invalid query in URL Builder Error")

except NoResultFound as nrf:
    raise Exception(
        "No URL was found for filters integration - {} domain - {} url - {}".format(integration_filter,
                                                                                    domain_filter, url_filter))

except MultipleResultsFound as mrf:
    raise Exception(
        "Multiple Results was found for filters integration - {} domain - {} url - {}".format(
            integration_filter, domain_filter, url_filter))

except Exception as e:
    raise Exception("Error while fetching records in URL Builder {}")

finally:
    session.close()