1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303
|
import warnings
from collections import OrderedDict
from typing import Type # noqa: F401
from django.db.models import Model # noqa: F401
import graphene
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 .settings import graphene_settings
from .utils import (
DJANGO_FILTER_INSTALLED,
camelize,
get_model_fields,
is_valid_django_model,
)
ALL_FIELDS = "__all__"
def construct_fields(
model, registry, only_fields, exclude_fields, convert_choices_to_enum=None
):
_model_fields = get_model_fields(model)
fields = OrderedDict()
for name, field in _model_fields:
is_not_in_only = (
only_fields is not None
and only_fields != ALL_FIELDS
and name not in only_fields
)
# is_already_created = name in options.fields
is_excluded = (
exclude_fields is not None and 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
_convert_choices_to_enum = convert_choices_to_enum
if isinstance(_convert_choices_to_enum, list):
# then `convert_choices_to_enum` is a list of field names to convert
if name in _convert_choices_to_enum:
_convert_choices_to_enum = True
else:
_convert_choices_to_enum = False
converted = convert_django_field_with_choices(
field, registry, convert_choices_to_enum=_convert_choices_to_enum
)
fields[name] = converted
return fields
def validate_fields(type_, model, fields, only_fields, exclude_fields):
# Validate the given fields against the model's fields and custom fields
all_field_names = set(fields.keys())
only_fields = only_fields if only_fields is not ALL_FIELDS else ()
for name in only_fields or ():
if name in all_field_names:
continue
if hasattr(model, name):
warnings.warn(
(
'Field name "{field_name}" matches an attribute on Django model "{app_label}.{object_name}" '
"but it's not a model field so Graphene cannot determine what type it should be. "
'Either define the type of the field on DjangoObjectType "{type_}" or remove it from the "fields" list.'
).format(
field_name=name,
app_label=model._meta.app_label,
object_name=model._meta.object_name,
type_=type_,
)
)
else:
warnings.warn(
(
'Field name "{field_name}" doesn\'t exist on Django model "{app_label}.{object_name}". '
'Consider removing the field from the "fields" list of DjangoObjectType "{type_}" because it has no effect.'
).format(
field_name=name,
app_label=model._meta.app_label,
object_name=model._meta.object_name,
type_=type_,
)
)
# Validate exclude fields
for name in exclude_fields or ():
if name in all_field_names:
# Field is a custom field
warnings.warn(
f'Excluding the custom field "{name}" on DjangoObjectType "{type_}" has no effect. '
'Either remove the custom field or remove the field from the "exclude" list.'
)
else:
if not hasattr(model, name):
warnings.warn(
(
'Django model "{app_label}.{object_name}" does not have a field or attribute named "{field_name}". '
'Consider removing the field from the "exclude" list of DjangoObjectType "{type_}" because it has no effect'
).format(
field_name=name,
app_label=model._meta.app_label,
object_name=model._meta.object_name,
type_=type_,
)
)
class DjangoObjectTypeOptions(ObjectTypeOptions):
model = None # type: 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=None, # deprecated in favour of `fields`
fields=None,
exclude_fields=None, # deprecated in favour of `exclude`
exclude=None,
filter_fields=None,
filterset_class=None,
connection=None,
connection_class=None,
use_connection=None,
interfaces=(),
convert_choices_to_enum=None,
_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), (
f"The attribute registry in {cls.__name__} needs to be an instance of "
f'Registry, received "{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"
)
assert not (fields and exclude), (
"Cannot set both 'fields' and 'exclude' options on "
f"DjangoObjectType {cls.__name__}."
)
# Alias only_fields -> fields
if only_fields and fields:
raise Exception("Can't set both only_fields and fields")
if only_fields:
warnings.warn(
"Defining `only_fields` is deprecated in favour of `fields`.",
DeprecationWarning,
stacklevel=2,
)
fields = only_fields
if fields and fields != ALL_FIELDS and not isinstance(fields, (list, tuple)):
raise TypeError(
'The `fields` option must be a list or tuple or "__all__". '
"Got %s." % type(fields).__name__
)
# Alias exclude_fields -> exclude
if exclude_fields and exclude:
raise Exception("Can't set both exclude_fields and exclude")
if exclude_fields:
warnings.warn(
"Defining `exclude_fields` is deprecated in favour of `exclude`.",
DeprecationWarning,
stacklevel=2,
)
exclude = exclude_fields
if exclude and not isinstance(exclude, (list, tuple)):
raise TypeError(
"The `exclude` option must be a list or tuple. Got %s."
% type(exclude).__name__
)
if fields is None and exclude is None:
warnings.warn(
"Creating a DjangoObjectType without either the `fields` "
"or the `exclude` option is deprecated. Add an explicit `fields "
f"= '__all__'` option on DjangoObjectType {cls.__name__} to use all "
"fields",
DeprecationWarning,
stacklevel=2,
)
django_fields = yank_fields_from_attrs(
construct_fields(model, registry, fields, exclude, convert_choices_to_enum),
_as=graphene.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(options.get("name") or cls.__name__), node=cls
)
if connection is not None:
assert issubclass(
connection, Connection
), f"The connection must be a Connection. Received {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
_meta.convert_choices_to_enum = convert_choices_to_enum
super().__init_subclass_with_meta__(
_meta=_meta, interfaces=interfaces, **options
)
# Validate fields
validate_fields(cls, model, _meta.fields, fields, exclude)
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, cls):
return True
if not is_valid_django_model(root.__class__):
raise Exception(f'Received incompatible instance "{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)
@classmethod
def from_errors(cls, errors):
data = camelize(errors) if graphene_settings.CAMELCASE_ERRORS else errors
return [cls(field=key, messages=value) for key, value in data.items()]
|