in metaflow/plugins/events_decorator.py [0:0]
def format_deploytime_value(self):
new_triggers = []
for trigger in self.triggers:
# Case where trigger is a function that returns a list of events
# Need to do this bc we need to iterate over list later
if isinstance(trigger, DeployTimeField):
evaluated_trigger = deploy_time_eval(trigger)
if isinstance(evaluated_trigger, dict):
trigger = evaluated_trigger
elif isinstance(evaluated_trigger, str):
trigger = {"name": evaluated_trigger}
if isinstance(evaluated_trigger, list):
for trig in evaluated_trigger:
if is_stringish(trig):
new_triggers.append({"name": trig})
else: # dict or another deploytimefield
new_triggers.append(trig)
else:
new_triggers.append(trigger)
else:
new_triggers.append(trigger)
self.triggers = new_triggers
for trigger in self.triggers:
old_trigger = trigger
trigger_params = trigger.get("parameters", {})
# Case where param is a function (can return list or dict)
if isinstance(trigger_params, DeployTimeField):
trigger_params = deploy_time_eval(trigger_params)
# If params is a list of strings, convert to dict with same key and value
if isinstance(trigger_params, (list, tuple)):
new_trigger_params = {}
for mapping in trigger_params:
if is_stringish(mapping) or callable(mapping):
new_trigger_params[mapping] = mapping
elif callable(mapping) and not isinstance(mapping, DeployTimeField):
mapping = DeployTimeField(
"parameter_val", str, None, mapping, False
)
new_trigger_params[mapping] = mapping
elif isinstance(mapping, (list, tuple)) and len(mapping) == 2:
if callable(mapping[0]) and not isinstance(
mapping[0], DeployTimeField
):
mapping[0] = DeployTimeField(
"parameter_val",
str,
None,
mapping[1],
False,
)
if callable(mapping[1]) and not isinstance(
mapping[1], DeployTimeField
):
mapping[1] = DeployTimeField(
"parameter_val",
str,
None,
mapping[1],
False,
)
new_trigger_params[mapping[0]] = mapping[1]
else:
raise MetaflowException(
"The *parameters* attribute for event '%s' is invalid. "
"It should be a list/tuple of strings and lists/tuples "
"of size 2" % self.attributes["event"]["name"]
)
trigger_params = new_trigger_params
trigger["parameters"] = trigger_params
trigger_name = trigger.get("name")
# Case where just the name is a function (always a str)
if isinstance(trigger_name, DeployTimeField):
trigger_name = deploy_time_eval(trigger_name)
trigger["name"] = trigger_name
# Third layer
# {name:, parameters:[func, ..., ...]}
# {name:, parameters:{func : func2}}
for trigger in self.triggers:
old_trigger = trigger
trigger_params = trigger.get("parameters", {})
new_trigger_params = {}
for key, value in trigger_params.items():
if isinstance(value, DeployTimeField) and key is value:
evaluated_param = deploy_time_eval(value)
new_trigger_params[evaluated_param] = evaluated_param
elif isinstance(value, DeployTimeField):
new_trigger_params[key] = deploy_time_eval(value)
elif isinstance(key, DeployTimeField):
new_trigger_params[deploy_time_eval(key)] = value
else:
new_trigger_params[key] = value
trigger["parameters"] = new_trigger_params
self.triggers[self.triggers.index(old_trigger)] = trigger