-
Notifications
You must be signed in to change notification settings - Fork 10
/
entity.py
136 lines (108 loc) · 4.27 KB
/
entity.py
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
from __future__ import annotations
from typing import Any, Callable, Dict
from graphene import List, NonNull, Union
from graphene.types.schema import Schema
from graphene.types.schema import TypeMap
from .types import _Any
from .utils import (
field_name_to_type_attribute,
check_fields_exist_on_type,
is_valid_compound_key,
)
import collections.abc
def update(d, u):
for k, v in u.items():
if isinstance(v, collections.abc.Mapping):
d[k] = update(d.get(k, {}), v)
else:
d[k] = v
return d
def get_entities(schema: Schema) -> Dict[str, Any]:
"""
Find all the entities from the type map.
They can be easily distinguished from the other type as
the `@key` and `@extend` decorators adds a `_sdl` attribute to them.
"""
type_map: TypeMap = schema.graphql_schema.type_map
entities = {}
for type_name, type_ in type_map.items():
if not hasattr(type_, "graphene_type"):
continue
if getattr(type_.graphene_type, "_keys", None):
entities[type_name] = type_.graphene_type
# Validation for compound keys
key_str = " ".join(type_.graphene_type._keys)
type_name = type_.graphene_type._meta.name
if "{" in key_str: # checking for subselection to identify compound key
assert is_valid_compound_key(
type_name, key_str, schema
), f'Invalid compound key definition for type "{type_name}"'
return entities
def get_entity_cls(entities: Dict[str, Any]) -> Union:
"""
Create _Entity type which is a union of all the entities types.
"""
class _Entity(Union):
class Meta:
types = tuple(entities.values())
return _Entity
def get_entity_query(schema: Schema):
"""
Create Entity query.
"""
entities_dict = get_entities(schema)
if not entities_dict:
return
entity_type = get_entity_cls(entities_dict)
class EntityQuery:
entities = List(
entity_type,
name="_entities",
representations=NonNull(List(NonNull(_Any))),
required=True,
)
def resolve_entities(self, info, representations):
entities = []
for representation in representations:
type_ = schema.graphql_schema.get_type(representation["__typename"])
model = type_.graphene_type
model_arguments = representation.copy()
model_arguments.pop("__typename")
if schema.auto_camelcase:
get_model_attr = field_name_to_type_attribute(schema, model)
model_arguments = {
get_model_attr(k): v for k, v in model_arguments.items()
}
model_instance = model(**model_arguments)
resolver = getattr(
model, "_%s__resolve_reference" % model.__name__, None
) or getattr(model, "_resolve_reference", None)
if resolver:
model_instance = resolver(model_instance, info)
entities.append(model_instance)
return entities
return EntityQuery
def key(fields: str, resolvable: bool = True) -> Callable:
"""
Take as input a field that should be used as key for that entity.
See specification: https://www.apollographql.com/docs/federation/federation-spec/#key
"""
def decorator(type_):
# Check the provided fields actually exist on the Type.
if " " not in fields:
assert (
fields in type_._meta.fields
), f'Field "{fields}" does not exist on type "{type_._meta.name}"'
if "{" not in fields:
# Skip valid fields check if the key is a compound key. The validation for compound keys
# is done on calling get_entities()
fields_set = set(fields.split(" "))
assert check_fields_exist_on_type(
fields=fields_set, type_=type_
), f'Field "{fields}" does not exist on type "{type_._meta.name}"'
keys = getattr(type_, "_keys", [])
keys.append(fields)
setattr(type_, "_keys", keys)
setattr(type_, "_resolvable", resolvable)
return type_
return decorator