Support different types of search filters (SQLAlchemy) #558
Description
Filter Version: SQLAlchemy
Currently, the search query parameter translates to a SQL query with this form:
SELECT * FROM example WHERE example.name ILIKE '%test_string%' OR example.description ILIKE '%test_string%';For my use case, I prefer to create trigram indexes and search on that index with this form:
SELECT * FROM example WHERE (example.name || ' ' || coalesce(example.description, '')) ILIKE '%test_string%';Additionally, I will want to search using tsvectors/tsqueries in the near future. I propose we create a search Enum, specific to the sqlalchemy filter code, that will have the values ILIKE_OR (current version), TRIGRAM, TSVECTOR (maybe a few others). It's actually quite easy to implement and I am running it in my project. I would be happy to put together a PR if there is interest in search customization like this.
I would also take care of coalescing nullable columns to ensure our search results are accurate.
Adding just the TRIGRAM search looks roughly like this but my PR would be much cleaner:
def get_ilike_field(model: Any, field: Any) -> Any:
column = getattr(model, field)
if column.nullable:
return func.coalesce(column, "")
return column
def filter(self, query: Union[Query, Select]):
....
if field_name == self.Constants.search_field_name and hasattr(self.Constants, "search_model_fields"):
if hasattr(self.Constants, "search_type") and self.Constants.search_type == "trigram":
search_value = None
for i, field in enumerate(self.Constants.search_model_fields):
ilike_value = get_ilike_field(self.Constants.model, field)
search_value = ilike_value if search_value is None else search_value + ilike_value
if i < len(self.Constants.search_model_fields) - 1:
search_value += " "
query = query.filter(ilike_op(search_value, f"%{value.lower()}%"))
else: # current default search
search_filters = [
getattr(self.Constants.model, field).ilike(f"%{value}%")
for field in self.Constants.search_model_fields
]
query = query.filter(or_(*search_filters))
else:
model_field = getattr(self.Constants.model, field_name)
query = query.filter(getattr(model_field, operator)(value))
return query