Custom filter backends¶
Filter backends can:
- Add new
graphene
input types to the (query) schema. - Allow you to request additional information by adding new
graphene
fields to the schema. - Alter current queryset.
- Alter slice, add additional information next to
pageInfo
andedges
, such asfacets
, for example.
Let’s learn by example on the SourceFilterBackend
which allows us
to apply source
query to the current search queryset.
import enum
import graphene
from graphen_elastic.filter_backends.base import BaseBackend
from graphen_elastic.constants import DYNAMIC_CLASS_NAME_PREFIX
class SourceFilterBackend(BaseBackend):
"""Source filter backend."""
prefix = 'source' # Name of the GraphQL query filter
has_query_fields = True # Indicates whether backend has own filtering fields
# The ``source_fields`` is the config options that we set on the
# ``Post`` object type. In this case - absolutely optional.
@property
def source_fields(self):
"""Source filter fields."""
return getattr(
self.connection_field.type._meta.node._meta,
'filter_backend_options',
{}
).get('source_fields', {})
# This is where we dynamically create GraphQL filter fields for this
# backend.
def get_backend_filtering_fields(self, items, is_filterable_func, get_type_func):
"""Construct backend fields.
:param items:
:param is_filterable_func:
:param get_type_func:
:return:
"""
_keys = list(
self.connection_field.type._meta.node._meta.fields.keys()
)
_keys.remove('_id')
params = zip(_keys, _keys)
return {
self.prefix: graphene.Argument(
graphene.List(
graphene.Enum.from_enum(
enum.Enum(
"{}{}{}BackendEnum".format(
DYNAMIC_CLASS_NAME_PREFIX,
self.prefix.title(),
self.connection_field.type.__name__
),
params
)
)
)
)
}
# Some data normalisation.
def prepare_source_fields(self):
"""Prepare source fields.
Possible structures:
source_fields = ["title"]
Or:
search_fields = ["title", "author.*"]
Or:
source = {
"includes": ["title", "author.*"],
"excludes": [ "*.description" ]
}
:return: Filtering options.
:rtype: dict
"""
source_args = dict(self.args).get(self.prefix, [])
source_fields = dict(self.source_fields)
if source_args:
return source_args
return source_fields
# This is where the queryset is being altered.
def filter(self, queryset):
"""Filter.
:param queryset:
:return:
"""
source_fields = self.prepare_source_fields()
if source_fields:
queryset = queryset.source(source_fields)
return queryset