Enhancement: Add StatementFilters for relationships & association object patterns

[ftsartek]

Summary

Consider adding additional filters that allow filtering on relationship attributes and association object patterns. Happy to make a PR based off the examples given here.

I've had a couple brief discussions about this with @cofin, so I'm making this as a feature request so we're not cluttering up the discord with it.

Basic Example

A RelationshipFilter could be relatively simple (with some smarter logic, it could even be integrated into an existing filter - though being explicit may be preferable so users are always aware they're working with relationships. I have been using this example for my relationships:

@dataclass
class RelationshipFilter(StatementFilter):
    relationship_name: str
    field_name: str
    value: str | None

    def append_to_statement(
        self,
        statement: Select[tuple[ModelT]],
        model: type[ModelT],
    ) -> Select[tuple[ModelT]]:
        relationship = self._get_relationship_attr(model, self.relationship_name)

        if self.value is None:
            return statement

        filter_condition = self._build_filter_condition(relationship)

        if relationship.property.uselist:
            # One-to-Many relationship
            return statement.where(relationship.any(filter_condition))
        # Many-to-One or One-to-One relationship
        return statement.where(relationship.has(filter_condition))

    def _build_filter_condition(
        self,
        relationship: InstrumentedAttribute[any],
    ) -> BinaryExpression:
        related_model = relationship.property.mapper.class_
        field = getattr(related_model, self.field_name)
        return field.ilike(f"%{self.value}%")

    @staticmethod
    def _get_relationship_attr(
        model: any,
        key: str | InstrumentedAttribute[any],
    ) -> InstrumentedAttribute[any]:
        attr = StatementFilter._get_instrumented_attr(model, key)
        if not isinstance(attr.property, RelationshipProperty):
            raise ValueError(f"{key} is not a relationship attribute")
        return attr

And this example for Association Object patterns. There's likely a world where these two filters can be combined, but again being explicit is probably preferable.

@dataclass
class AssociationObjectFilter(StatementFilter):
    relationship_name: str
    association_remote_relationship: str
    remote_field_name: str
    remote_value: str | None
    association_field: str | None = None
    association_value: str | None = None

    def append_to_statement(
        self,
        statement: Select[tuple[ModelT]],
        model: type[ModelT],
    ) -> Select[tuple[ModelT]]:
        relationship = self._get_relationship_attr(model, self.relationship_name)

        if self.remote_value is None and (
            self.association_value is None or self.association_field is None
        ):
            return statement

        filter_condition = self._build_filter_condition(relationship)
        return statement.where(relationship.any(filter_condition))

    def _build_filter_condition(
        self,
        relationship: InstrumentedAttribute[any],
    ) -> BinaryExpression:
        conditions = []

        association_class = relationship.property.mapper.class_
        remote_relationship = getattr(
            association_class,
            self.association_remote_relationship,
        )
        remote_class = remote_relationship.property.mapper.class_

        if self.remote_value is not None:
            remote_field = getattr(remote_class, self.remote_field_name)
            conditions.append(
                remote_relationship.has(remote_field.ilike(f"%{self.remote_value}%")),
            )

        if self.association_field and self.association_value is not None:
            association_attr = getattr(association_class, self.association_field)
            conditions.append(association_attr == self.association_value)

        return and_(true(), *conditions)

    @staticmethod
    def _get_relationship_attr(
        model: any,
        key: str | InstrumentedAttribute[any],
    ) -> InstrumentedAttribute[any]:
        attr = StatementFilter._get_instrumented_attr(model, key)
        if not isinstance(attr.property, RelationshipProperty):
            raise ValueError(f"{key} is not a relationship attribute")
        return attr

Drawbacks and Impact

These open up another route to hitting relationship join method issues (e.g. these will currently error out on a "noload" relationship, and of course a "raiseload" relationship). They should probably check the join status in the current object, particularly since they'll often be used in async contexts where lazy loading is a no-go.

The examples posted are also probably in need of some optimisation.

Unresolved questions

No response