SQLAlchemy query db with filter for all tables

Question

I have SQLAlchemy models on top of the MySQL db. I need to query almost all models (string or text fields) and find everything that contains a specific substring. And also, apply common filtering like object_type=type1. For exsmple:

class Model1(Model):
    name = Column(String(100), nullable=False, unique=True)
    version = Column(String(100))
    description = Column(String(100))
    updated_at = Column(TIMESTAMP(timezone=True))
    # other fields

class Model2(Model):
    name = Column(String(100), nullable=False, unique=True)
    version = Column(String(100))
    description = Column(String(100))
    updated_at = Column(TIMESTAMP(timezone=True))
    # other fields

class Model3(Model):
    name = Column(String(100), nullable=False, unique=True)
    version = Column(String(100))
    description = Column(String(100))
    updated_at = Column(TIMESTAMP(timezone=True))
    # other fields

And then do query something like:

db.query(
    Model1.any_of_all_columns.contains('sub_string') or
    Model2.any_of_all_columns.contains('sub_string') or
    Model3.any_of_all_columns.contains('sub_string')
).all()

Is it possible to build such an ORM query in one SQL to the db and dynamically add Model(table) names and columns?

Answer 1

For applying common filtering for all the columns, you can subscribe to sqlachemy events as following:

@event.listens_for(Query, "before_compile", retval=True)
def before_compile(query):

for ent in query.column_descriptions:
    entity = ent['entity']
    if entity is None:
        continue


    inspect_entity_for_mapper = inspect(ent['entity'])
    mapper = getattr(inspect_entity_for_mapper, 'mapper', None)

    if mapper and has_tenant_id:

        query = query.enable_assertions(False).filter(
            ent['entity’].object == object)

return query

This function will be called whenever you do Model.query() and add filter for your object.

Answer 2

I eventually gave up and did one big loop in which I make a separate request for each model:

    from sqlalchemy import or_

    def db_search(self, model, q, object_ids=None, status=None, m2m_ids=None):
    """
        Build a query to the db for given model using 'q' search substring
        and filter it by object ids, its status and m2m related model.
        :param model: a model object which columns will be used for search.
        :param q: the query substring we are trying to find in all
            string/text columns of the model.
        :param object_ids: list of ids we want to include in the search.
            If the list is empty, the search query will return 0 results.
            If object_ids is None, we will ignore this filter.
        :param status: name of object status.
        :param m2m_ids: list of many-to-many related object ids.
        :return: sqlalchemy query result.
    """

    # Filter out private columns and not string/text columns
    string_text_columns = [
        column.name for column in model.__table__.columns if
        isinstance(column.type, (db.String, db.Text))
        and column.name not in PRIVATE_COLUMN_NAMES
    ]

    # Find only enum ForeignKey columns
    foreign_key_columns = [
        column.name for column in model.__table__.columns if
        column.name.endswith("_id") and column.name in ENUM_OBJECTS
        )
    ]

    query_result = model.query

    # Search in all string/text columns for the required query
    # as % LIKE %
    if q:
        query_result = query_result.join(
            # Join related enum tables for being able to search in
            *[enum_tables_to_model_map[col]["model_name"] for col in
              foreign_key_columns]
        ).filter(
            or_(
                # Search 'q' substring in all string/text columns
                *[
                    getattr(model, col_name).like(f"%{q}%")
                    for col_name in string_text_columns
                ],
                # Search 'q' substring in the enum tables
                *[
                    enum_tables_to_model_map[col]["model_field"]
                    .like(f"%{q}%") for col in foreign_key_columns
                ]
            )
        )

    # Apply filter by object ids if given and it's not None.
    # If the object ids filter exist but it's empty, we should
    # return an empty result
    if object_ids is not None:
        query_result = query_result.filter(model.id.in_(object_ids))

    # Apply filter by status if given and if the model has the status
    # column
    if status and 'status_id' in model.__table__.columns:
        query_result = query_result.filter(model.status_id == status.id)
    if m2m_ids:
        query_result = query_result.filter(
            model.labels.any(Label.id.in_(m2m_ids)))

    return query_result.all()

And call it:

        result = {}
        for model in db.Model._decl_class_registry.values():
        # Search only in the public tables
        # sqlalchemy.ext.declarative.clsregistry._ModuleMarker object
        # located in the _decl_class_registry that is why we check
        # instance type and whether it is subclass of the db.Model
        if isinstance(model, type) and issubclass(model, db.Model) \
                and model.__name__ in PUBLIC_MODEL_NAMES:

            query_result = self.db_search(
                model, q, object_ids.get(model.__name__), status=status,
                m2m_ids=m2m_ids)

            result[model.__tablename__] = query_result

This is far from the best solution, but it works for me.