fastapi-filter icon indicating copy to clipboard operation
fastapi-filter copied to clipboard

Ignoring order_by for related fields in existing filters [SQLAlchemy]

Open espdev opened this issue 2 years ago • 6 comments

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)

espdev avatar Jul 10 '23 19:07 espdev

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

ytjohn avatar Jul 23 '23 23:07 ytjohn

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.

github-actions[bot] avatar Jan 12 '24 01:01 github-actions[bot]

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.

github-actions[bot] avatar Apr 09 '24 01:04 github-actions[bot]