graphql-python / graphene-sqlalchemy

Graphene SQLAlchemy integration
http://docs.graphene-python.org/projects/sqlalchemy/en/latest/
MIT License
980 stars 226 forks source link

Allow a custom filter class with the purpose of using all the base filters, and adding sqlalchemy-filter esk filters #407

Closed adiberk closed 6 months ago

adiberk commented 6 months ago

So a pretty important use case at my company is the ability to add custom filters that aren't field specific Here is an example use case using the below hack discussed

class UserNode(SQLAlchemyObjectType):
    class Meta:
        model = User
        interfaces = (LevNode,)
        filter = UserFilter

class UserFilter(GrapheneSQLAlchemyFilter):
    use_has_contact = graphene.Boolean()
    is_valid = graphene.Boolean()

    @staticmethod
    def user_in_filter(info: LevResolveInfo, query: Query, value: bool) -> Query:
        return query.join(Contact).filter(Contact.id.is_not(None))

    @staticmethod
    def is_valid_filter(info: LevResolveInfo, query: Query, value: bool) -> ColumnElement:
        if value:
            return User.deleted_at.is_(None)
        else:
            return User.deleted_at.is_not(None)

Step 1. Update BaseTypeFilter class to allow for "filter" as a _meta field. We get all the custom filter functions from the classes that extend GrapheneSQLAlchemyFilter. We ensure those functions contain the correct variables. and then add the fields to the filter fields list.

class GrapheneSQLAlchemyFilter(graphene.InputObjectType):
    pass

class BaseTypeFilter(graphene.InputObjectType):
    @classmethod
    def __init_subclass_with_meta__(
        cls, filter_fields=None, model=None, _meta=None, custom_filter_class=None, **options
    ):
        from graphene_sqlalchemy.converter import convert_sqlalchemy_type

        # Init meta options class if it doesn't exist already
        if not _meta:
            _meta = InputObjectTypeOptions(cls)
        _meta.filter_class = custom_filter_class
        logic_functions = _get_functions_by_regex(".+_logic$", "_logic$", cls)
        custom_filter_fields = {}
        if custom_filter_class and issubclass(custom_filter_class, GrapheneSQLAlchemyFilter):
            custom_filter_fields = yank_fields_from_attrs(custom_filter_class.__dict__, _as=graphene.InputField)
            functions = dict(_get_functions_by_regex(".+_filter$", "_filter$", custom_filter_class))
            for field_name in custom_filter_fields.keys():
                assert functions.get(field_name), f"Custom filter field {field_name} must have a corresponding filter method"
                annotations = functions.get(field_name)
                assert annotations.get('info'), "Each custom filter method must have an info field with valid type annotations"
                assert annotations.get('query'), "Each custom filter method must have a query field with valid type annotations"
                assert annotations.get('value'), "Each custom filter method must have a value field with valid type annotations"
        new_filter_fields = custom_filter_fields
       ..........

Then override the execute_filters method. We have it accept a "resolve_info" or "info" so that we can pass those to the custom filter functions

    @classmethod
    def execute_filters(
        cls, query, filter_dict: Dict[str, Any], model_alias=None, info=None
    ) -> Tuple[Query, List[Any]]:
        model = cls._meta.model
        .....
        # Here we first check if this is input field that isn't a model_attr and is part of the filter_class (we set that on the meta earlier)
        else:
                # Allow custom filter class to be used for custom filtering over 
                if not hasattr(input_field, "model_attr") and cls._meta.filter_class:
                    clause = getattr(cls._meta.filter_class, field + "_filter")(info, query, field_filters)
                    if isinstance(clause, tuple):
                        query, clause = clause
                    elif isinstance(clause, Query):
                        query = clause
                        continue
                    clauses.append(clause)
                else:
                    model_field = getattr(model, input_field.model_attr or field)

Update SQLAlchemy base to accept a "filter" field

class SQLAlchemyObjectTypeOptions(ObjectTypeOptions):
    .....
    filter = None
github-actions[bot] commented 2 weeks ago

This issue has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related topics referencing this issue.