-
-
Notifications
You must be signed in to change notification settings - Fork 2k
/
dependencies.py
368 lines (281 loc) · 12.6 KB
/
dependencies.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
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
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
import json
from dash.development.base_component import Component
from ._validate import validate_callback
from ._grouping import flatten_grouping, make_grouping_by_index
class _Wildcard: # pylint: disable=too-few-public-methods
def __init__(self, name):
self._name = name
def __str__(self):
return self._name
def __repr__(self):
return f"<{self}>"
def to_json(self):
# used in serializing wildcards - arrays are not allowed as
# id values, so make the wildcards look like length-1 arrays.
return f'["{self._name}"]'
MATCH = _Wildcard("MATCH")
ALL = _Wildcard("ALL")
ALLSMALLER = _Wildcard("ALLSMALLER")
class DashDependency: # pylint: disable=too-few-public-methods
def __init__(self, component_id, component_property):
if isinstance(component_id, Component):
self.component_id = component_id._set_random_id()
else:
self.component_id = component_id
self.component_property = component_property
def __str__(self):
return f"{self.component_id_str()}.{self.component_property}"
def __repr__(self):
return f"<{self.__class__.__name__} `{self}`>"
def component_id_str(self):
i = self.component_id
def _dump(v):
return json.dumps(v, sort_keys=True, separators=(",", ":"))
def _json(k, v):
vstr = v.to_json() if hasattr(v, "to_json") else json.dumps(v)
return f"{json.dumps(k)}:{vstr}"
if isinstance(i, dict):
return "{" + ",".join(_json(k, i[k]) for k in sorted(i)) + "}"
return i
def to_dict(self):
return {"id": self.component_id_str(), "property": self.component_property}
def __eq__(self, other):
"""
We use "==" to denote two deps that refer to the same prop on
the same component. In the case of wildcard deps, this means
the same prop on *at least one* of the same components.
"""
return (
isinstance(other, DashDependency)
and self.component_property == other.component_property
and self._id_matches(other)
)
def _id_matches(self, other):
my_id = self.component_id
other_id = other.component_id
self_dict = isinstance(my_id, dict)
other_dict = isinstance(other_id, dict)
if self_dict != other_dict:
return False
if self_dict:
if set(my_id.keys()) != set(other_id.keys()):
return False
for k, v in my_id.items():
other_v = other_id[k]
if v == other_v:
continue
v_wild = isinstance(v, _Wildcard)
other_wild = isinstance(other_v, _Wildcard)
if v_wild or other_wild:
if not (v_wild and other_wild):
continue # one wild, one not
if v is ALL or other_v is ALL:
continue # either ALL
if v is MATCH or other_v is MATCH:
return False # one MATCH, one ALLSMALLER
else:
return False
return True
# both strings
return my_id == other_id
def __hash__(self):
return hash(str(self))
def has_wildcard(self):
"""
Return true if id contains a wildcard (MATCH, ALL, or ALLSMALLER)
"""
if isinstance(self.component_id, dict):
for v in self.component_id.values():
if isinstance(v, _Wildcard):
return True
return False
class Output(DashDependency): # pylint: disable=too-few-public-methods
"""Output of a callback."""
allowed_wildcards = (MATCH, ALL)
class Input(DashDependency): # pylint: disable=too-few-public-methods
"""Input of callback: trigger an update when it is updated."""
allowed_wildcards = (MATCH, ALL, ALLSMALLER)
class State(DashDependency): # pylint: disable=too-few-public-methods
"""Use the value of a State in a callback but don't trigger updates."""
allowed_wildcards = (MATCH, ALL, ALLSMALLER)
class ClientsideFunction: # pylint: disable=too-few-public-methods
def __init__(self, namespace=None, function_name=None):
if namespace.startswith("_dashprivate_"):
raise ValueError("Namespaces cannot start with '_dashprivate_'.")
if namespace in ["PreventUpdate", "no_update"]:
raise ValueError(
f'"{namespace}" is a forbidden namespace in dash_clientside.'
)
self.namespace = namespace
self.function_name = function_name
def __repr__(self):
return f"ClientsideFunction({self.namespace}, {self.function_name})"
def extract_grouped_output_callback_args(args, kwargs):
if "output" in kwargs:
parameters = kwargs["output"]
# Normalize list/tuple of multiple positional outputs to a tuple
if isinstance(parameters, (list, tuple)):
parameters = list(parameters)
# Make sure dependency grouping contains only Output objects
for dep in flatten_grouping(parameters):
if not isinstance(dep, Output):
raise ValueError(
f"Invalid value provided where an Output dependency "
f"object was expected: {dep}"
)
return parameters
parameters = []
while args:
next_deps = flatten_grouping(args[0])
if all(isinstance(d, Output) for d in next_deps):
parameters.append(args.pop(0))
else:
break
return parameters
def extract_grouped_input_state_callback_args_from_kwargs(kwargs):
input_parameters = kwargs["inputs"]
if isinstance(input_parameters, DashDependency):
input_parameters = [input_parameters]
state_parameters = kwargs.get("state", None)
if isinstance(state_parameters, DashDependency):
state_parameters = [state_parameters]
if isinstance(input_parameters, dict):
# Wrapped function will be called with named keyword arguments
if state_parameters:
if not isinstance(state_parameters, dict):
raise ValueError(
"The input argument to app.callback was a dict, "
"but the state argument was not.\n"
"input and state arguments must have the same type"
)
# Merge into state dependencies
parameters = state_parameters
parameters.update(input_parameters)
else:
parameters = input_parameters
return parameters
if isinstance(input_parameters, (list, tuple)):
# Wrapped function will be called with positional arguments
parameters = list(input_parameters)
if state_parameters:
if not isinstance(state_parameters, (list, tuple)):
raise ValueError(
"The input argument to app.callback was a list, "
"but the state argument was not.\n"
"input and state arguments must have the same type"
)
parameters += list(state_parameters)
return parameters
raise ValueError(
"The input argument to app.callback may be a dict, list, or tuple,\n"
f"but received value of type {type(input_parameters)}"
)
def extract_grouped_input_state_callback_args_from_args(args):
# Collect input and state from args
parameters = []
while args:
next_deps = flatten_grouping(args[0])
if all(isinstance(d, (Input, State)) for d in next_deps):
parameters.append(args.pop(0))
else:
break
if len(parameters) == 1:
# Only one output grouping, return as-is
return parameters[0]
# Multiple output groupings, return wrap in tuple
return parameters
def extract_grouped_input_state_callback_args(args, kwargs):
if "inputs" in kwargs:
return extract_grouped_input_state_callback_args_from_kwargs(kwargs)
if "state" in kwargs:
# Not valid to provide state as kwarg without input as kwarg
raise ValueError(
"The state keyword argument may not be provided without "
"the input keyword argument"
)
return extract_grouped_input_state_callback_args_from_args(args)
def compute_input_state_grouping_indices(input_state_grouping):
# Flatten grouping of Input and State dependencies into a flat list
flat_deps = flatten_grouping(input_state_grouping)
# Split into separate flat lists of Input and State dependencies
flat_inputs = [dep for dep in flat_deps if isinstance(dep, Input)]
flat_state = [dep for dep in flat_deps if isinstance(dep, State)]
# For each entry in the grouping, compute the index into the
# concatenation of flat_inputs and flat_state
total_inputs = len(flat_inputs)
input_count = 0
state_count = 0
flat_inds = []
for dep in flat_deps:
if isinstance(dep, Input):
flat_inds.append(input_count)
input_count += 1
else:
flat_inds.append(total_inputs + state_count)
state_count += 1
# Reshape this flat list of indices to match the input grouping
grouping_inds = make_grouping_by_index(input_state_grouping, flat_inds)
return flat_inputs, flat_state, grouping_inds
def handle_grouped_callback_args(args, kwargs):
"""Split args into outputs, inputs and states"""
prevent_initial_call = kwargs.get("prevent_initial_call", None)
if prevent_initial_call is None and args and isinstance(args[-1], bool):
args, prevent_initial_call = args[:-1], args[-1]
# flatten args, to support the older syntax where outputs, inputs, and states
# each needed to be in their own list
flat_args = []
for arg in args:
flat_args += arg if isinstance(arg, (list, tuple)) else [arg]
outputs = extract_grouped_output_callback_args(flat_args, kwargs)
flat_outputs = flatten_grouping(outputs)
if isinstance(outputs, (list, tuple)) and len(outputs) == 1:
out0 = kwargs.get("output", args[0] if args else None)
if not isinstance(out0, (list, tuple)):
# unless it was explicitly provided as a list, a single output
# should be unwrapped. That ensures the return value of the
# callback is also not expected to be wrapped in a list.
outputs = outputs[0]
inputs_state = extract_grouped_input_state_callback_args(flat_args, kwargs)
flat_inputs, flat_state, input_state_indices = compute_input_state_grouping_indices(
inputs_state
)
types = Input, Output, State
validate_callback(flat_outputs, flat_inputs, flat_state, flat_args, types)
return outputs, flat_inputs, flat_state, input_state_indices, prevent_initial_call
def extract_callback_args(args, kwargs, name, type_):
"""Extract arguments for callback from a name and type"""
parameters = kwargs.get(name, [])
if parameters:
if not isinstance(parameters, (list, tuple)):
# accept a single item, not wrapped in a list, for any of the
# categories as a named arg (even though previously only output
# could be given unwrapped)
return [parameters]
else:
while args and isinstance(args[0], type_):
parameters.append(args.pop(0))
return parameters
def handle_callback_args(args, kwargs):
"""Split args into outputs, inputs and states"""
prevent_initial_call = kwargs.get("prevent_initial_call", None)
if prevent_initial_call is None and args and isinstance(args[-1], bool):
args, prevent_initial_call = args[:-1], args[-1]
# flatten args, to support the older syntax where outputs, inputs, and states
# each needed to be in their own list
flat_args = []
for arg in args:
flat_args += arg if isinstance(arg, (list, tuple)) else [arg]
outputs = extract_callback_args(flat_args, kwargs, "output", Output)
validate_outputs = outputs
if len(outputs) == 1:
out0 = kwargs.get("output", args[0] if args else None)
if not isinstance(out0, (list, tuple)):
# unless it was explicitly provided as a list, a single output
# should be unwrapped. That ensures the return value of the
# callback is also not expected to be wrapped in a list.
outputs = outputs[0]
inputs = extract_callback_args(flat_args, kwargs, "inputs", Input)
states = extract_callback_args(flat_args, kwargs, "state", State)
types = Input, Output, State
validate_callback(validate_outputs, inputs, states, flat_args, types)
return outputs, inputs, states, prevent_initial_call