graphene-django/graphene_django/types.py

116 lines
4.2 KiB
Python
Raw Normal View History

from collections import OrderedDict
from django.utils.functional import SimpleLazyObject
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)
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
2017-07-25 09:42:40 +03:00
is_excluded = name in exclude_fields # or is_already_created
2016-11-10 10:56:14 +03:00
# https://docs.djangoproject.com/en/1.10/ref/models/fields/#django.db.models.ForeignKey.related_query_name
2016-11-10 21:16:44 +03:00
is_no_backref = str(name).endswith('+')
2016-11-10 10:56:14 +03:00
if is_not_in_only or is_excluded or is_no_backref:
# We skip this field if we specify only_fields and is not
2016-11-10 10:56:14 +03:00
# 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 = ()
class DjangoObjectType(ObjectType):
@classmethod
def __init_subclass_with_meta__(cls, model=None, registry=None, skip_registry=False,
2017-07-25 09:42:40 +03:00
only_fields=(), exclude_fields=(), filter_fields=None, connection=None,
use_connection=None, interfaces=(), **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 not DJANGO_FILTER_INSTALLED and filter_fields:
raise Exception("Can only set filter_fields 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
connection = Connection.create_type('{}Connection'.format(cls.__name__), node=cls)
if connection is not None:
2017-07-25 09:42:40 +03:00
assert issubclass(connection, Connection), (
"The connection must be a Connection. Received {}"
).format(connection.__name__)
_meta = DjangoObjectTypeOptions(cls)
_meta.model = model
_meta.registry = registry
_meta.filter_fields = filter_fields
_meta.fields = django_fields
_meta.connection = connection
super(DjangoObjectType, cls).__init_subclass_with_meta__(_meta=_meta, interfaces=interfaces, **options)
2017-07-25 09:42:40 +03:00
if not skip_registry:
registry.register(cls)
2017-07-28 19:43:27 +03:00
def resolve_id(self, info):
return self.pk
@classmethod
2017-07-28 19:43:27 +03:00
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))
model = root._meta.model
return model == cls._meta.model
@classmethod
2017-07-28 19:43:27 +03:00
def get_node(cls, info, id):
try:
return cls._meta.model.objects.get(pk=id)
except cls._meta.model.DoesNotExist:
return None