SQLAlchemy query db，带有所有表的筛选器

Question

我在MySQL数据库之上有SQLAlchemy模型。我需要查询几乎所有模型(字符串或文本字段(并找到包含特定子字符串的所有内容。此外，应用常见的过滤，如 object_type=type1。例如：

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

然后查询类似以下内容：

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

是否可以在一个SQL中将这样的ORM查询构建到数据库中，并动态添加模型(表(名称和列？

Answer 1

要对所有列应用公共筛选，您可以按如下方式订阅 sqlachemy 事件：

@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

每当您执行 Model.query(( 并为对象添加过滤器时，都会调用此函数。

Answer 2

我最终放弃了，做了一个大循环，在这个循环中，我为每个模型提出一个单独的请求：

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

并称之为：

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

这远非最佳解决方案，但它对我有用。