Ignoring order_by for related fields in existing filters [SQLAlchemy]
Hello,
The current implementation of SQLAlchemy Filter ignores order_by in existing filters when it is used in related fields.
Here is a simple example:
from fastapi_filter import FilterDepends, with_prefix
from fastapi_filter.contrib.sqlalchemy import Filter
class UserFilter(Filter):
name: str | None
order_by: list[str] | None
class OrgUserFilter(Filter):
user: UserFilter | None = FilterDepends(with_prefix('user', UserFilter))
role: str | None
order_by: list[str] | None
In this case user__order_by will be ignored because the current implementation of Filter.sort method ignores ordering_values for related fields.
Also we cannot use prefix in this case:
order_by=user__created_at
We will get an error:
{
"detail": [
{
"loc": [
"order_by"
],
"msg": "user__created_at is not a valid ordering field.",
"type": "value_error"
}
]
}
How can we fix it?
A simple fix might look something like this:
class Filter(BaseFilterModel):
def sort(self, query: Union[Query, Select]):
# Apply sort for all related fields if it is applicable
for field_name, _ in self.filtering_fields:
field_value = getattr(self, field_name)
if isinstance(field_value, Filter):
query = field_value.sort(query)
if not self.ordering_values:
return query
for field_name in self.ordering_values:
direction = Filter.Direction.asc
if field_name.startswith("-"):
direction = Filter.Direction.desc
field_name = field_name.replace("-", "").replace("+", "")
order_by_field = getattr(self.Constants.model, field_name)
query = query.order_by(getattr(order_by_field, direction)())
return query
It seems the fix works fine for my cases. In my code I just make a child class from Filter and override sort method:
class MyFilter(Filter):
def sort(self, query):
for field_name, _ in self.filtering_fields:
field_value = getattr(self, field_name)
if isinstance(field_value, Filter):
query = field_value.sort(query)
return super().sort(query)
Hi, this worked pretty well for me, but with a few notes. I also have an alternative approach. I put the notes first that impact both your solution and mine.
First, in my joins, the generated SQL statement was appending _1 to the joined tables. But the sort was putting the actual tablename from the dbmodel. I use sqlalchemy's aliased method. I had to go back and set the name= parameter so that they would match what sort would expect.
UserTable` = aliased(dbmodel.User, name=dbmodel.User.__tablename__)
Second one is that if you follow the document's example to set a default sort on any of the related tables, then that default sort will get inserted. I had a default "-updated" for mine. I've reversed this, and now I'll only set a default at the api level.
@validator("order_by")
def sortable_fields(cls, value):
if value is None:
# return "-updated" # this got inserted before other sorting
return None
Third - I completely missed that you were using ?user__order_by=. This is on me. Once I realized that, I was able to get yours working. I was myself working down a path to keep everything in the ?order_by so that the order order by is more predictable. Keep in mind that your user__order_by=field3 will occur first, then if you have order_by=field1,field2.
In fact it's possible to support both methods.
I don't have a good fix for validate_order_by YET. What I have right now is a cheat.
@validator("*", allow_reuse=True, check_fields=False)
def validate_order_by(cls, value, values, field):
if field.name != cls.Constants.ordering_field_name:
return value
if not value:
return None
field_name_usages = defaultdict(list)
duplicated_field_names = set()
for field_name_with_direction in value:
field_name = field_name_with_direction.replace("-", "").replace("+", "")
if field_name.count("__") == 1: # cheap workaround
return value
But with that done, the following code works. For completeness, I left both methods in place.
# add in "prefix__order_by" related fields
# based on https://github.com/arthurio/fastapi-filter/issues/417
for field_name, _ in self.filtering_fields:
field_value = getattr(self, field_name)
if isinstance(field_value, Filter):
query = field_value.sort(query)
if not self.ordering_values:
return query
for field_name in self.ordering_values:
# clear the related fields first
# handle related fields in the format order_by="othertable__fieldname"
if field_name.count("__") == 1:
prefix, related = field_name.split("__")
field_value = getattr(self, prefix)
if isinstance(field_value, Filter):
ordering_field = field_value.Constants.ordering_field_name
setattr(field_value, ordering_field, [related])
# field_value.ordering_values = [related]
query = field_value.sort(query)
# go to next field_name
continue
This issue is stale because it has been open 60 days with no activity. Remove stale label or comment or this will be closed in 5 days.
This issue is stale because it has been open 60 days with no activity. Remove stale label or comment or this will be closed in 5 days.