Динамически отфильтровывать данные из sqlalchemy - PullRequest
Новая
       44

Динамически отфильтровывать данные из sqlalchemy

1 голос
/ 30 октября 2019

У меня есть эти модели 

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}

Теперь мне нужно отфильтровать данные на основе фильтров. Если предположить, что интеграционные_фильтры заданы, фильтруйте только данные интеграции. Если указаны Интеграция и Домен, отфильтруйте только домен и интеграцию, соединив их То же самое с URL и Доменом

Решения, которые я пробовал -

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)

1 Ответ

0 голосов
/ 30 октября 2019

Этот работал. Хотя было довольно просто, я не думал об этом в начале 

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()
...