graphene-django/graphene_django/types.py
Mel van Londen a9a8d672e9
Merge pull request #600 from sierreis/filterset-class
Add support for filterset_class meta parameter
2019-06-09 16:48:46 -07:00

166 lines
5.3 KiB
Python

import six
from collections import OrderedDict
from django.db.models import Model
from django.utils.functional import SimpleLazyObject
import graphene
from graphene import Field
from graphene.relay import Connection, Node
from graphene.types.objecttype import ObjectType, ObjectTypeOptions
from graphene.types.utils import yank_fields_from_attrs
from .converter import convert_django_field_with_choices
from .registry import Registry, get_global_registry
from .utils import DJANGO_FILTER_INSTALLED, get_model_fields, is_valid_django_model
if six.PY3:
from typing import Type
def construct_fields(model, registry, only_fields, exclude_fields):
_model_fields = get_model_fields(model)
fields = OrderedDict()
for name, field in _model_fields:
is_not_in_only = only_fields and name not in only_fields
# is_already_created = name in options.fields
is_excluded = name in exclude_fields # or is_already_created
# https://docs.djangoproject.com/en/1.10/ref/models/fields/#django.db.models.ForeignKey.related_query_name
is_no_backref = str(name).endswith("+")
if is_not_in_only or is_excluded or is_no_backref:
# We skip this field if we specify only_fields and is not
# in there. Or when we exclude this field in exclude_fields.
# Or when there is no back reference.
continue
converted = convert_django_field_with_choices(field, registry)
fields[name] = converted
return fields
class DjangoObjectTypeOptions(ObjectTypeOptions):
model = None # type: Model
registry = None # type: Registry
connection = None # type: Type[Connection]
filter_fields = ()
filterset_class = None
class DjangoObjectType(ObjectType):
@classmethod
def __init_subclass_with_meta__(
cls,
model=None,
registry=None,
skip_registry=False,
only_fields=(),
exclude_fields=(),
filter_fields=None,
filterset_class=None,
connection=None,
connection_class=None,
use_connection=None,
interfaces=(),
_meta=None,
**options
):
assert is_valid_django_model(model), (
'You need to pass a valid Django Model in {}.Meta, received "{}".'
).format(cls.__name__, model)
if not registry:
registry = get_global_registry()
assert isinstance(registry, Registry), (
"The attribute registry in {} needs to be an instance of "
'Registry, received "{}".'
).format(cls.__name__, registry)
if filter_fields and filterset_class:
raise Exception("Can't set both filter_fields and filterset_class")
if not DJANGO_FILTER_INSTALLED and (filter_fields or filterset_class):
raise Exception((
"Can only set filter_fields or filterset_class if "
"Django-Filter is installed"
))
django_fields = yank_fields_from_attrs(
construct_fields(model, registry, only_fields, exclude_fields), _as=Field
)
if use_connection is None and interfaces:
use_connection = any(
(issubclass(interface, Node) for interface in interfaces)
)
if use_connection and not connection:
# We create the connection automatically
if not connection_class:
connection_class = Connection
connection = connection_class.create_type(
"{}Connection".format(cls.__name__), node=cls
)
if connection is not None:
assert issubclass(connection, Connection), (
"The connection must be a Connection. Received {}"
).format(connection.__name__)
if not _meta:
_meta = DjangoObjectTypeOptions(cls)
_meta.model = model
_meta.registry = registry
_meta.filter_fields = filter_fields
_meta.filterset_class = filterset_class
_meta.fields = django_fields
_meta.connection = connection
super(DjangoObjectType, cls).__init_subclass_with_meta__(
_meta=_meta, interfaces=interfaces, **options
)
if not skip_registry:
registry.register(cls)
def resolve_id(self, info):
return self.pk
@classmethod
def is_type_of(cls, root, info):
if isinstance(root, SimpleLazyObject):
root._setup()
root = root._wrapped
if isinstance(root, cls):
return True
if not is_valid_django_model(type(root)):
raise Exception(('Received incompatible instance "{}".').format(root))
if cls._meta.model._meta.proxy:
model = root._meta.model
else:
model = root._meta.model._meta.concrete_model
return model == cls._meta.model
@classmethod
def get_queryset(cls, queryset, info):
return queryset
@classmethod
def get_node(cls, info, id):
queryset = cls.get_queryset(cls._meta.model.objects, info)
try:
return queryset.get(pk=id)
except cls._meta.model.DoesNotExist:
return None
class ErrorType(ObjectType):
field = graphene.String(required=True)
messages = graphene.List(graphene.NonNull(graphene.String), required=True)