in metaflow/flowspec.py [0:0]
def _process_config_decorators(cls, config_options, process_configs=True):
# Fast path for no user configurations
if not process_configs or (
not cls._flow_state.get(_FlowState.CONFIG_DECORATORS)
and all(len(step.config_decorators) == 0 for step in cls._steps)
):
# Process parameters to allow them to also use config values easily
for var, param in cls._get_parameters():
if param.IS_CONFIG_PARAMETER:
continue
param.init(not process_configs)
return None
debug.userconf_exec("Processing mutating step/flow decorators")
# We need to convert all the user configurations from DelayedEvaluationParameters
# to actual values so they can be used as is in the config decorators.
# We then reset them to be proper configs so they can be re-evaluated in
# _set_constants
to_reset_configs = []
cls._check_parameters(config_parameters=True)
for var, param in cls._get_parameters():
if not param.IS_CONFIG_PARAMETER:
continue
# Note that a config with no default and not required will be None
val = config_options.get(param.name.replace("-", "_").lower())
if isinstance(val, DelayedEvaluationParameter):
val = val()
# We store the value as well so that in _set_constants, we don't try
# to recompute (no guarantee that it is stable)
param._store_value(val)
to_reset_configs.append((var, param))
debug.userconf_exec("Setting config %s to %s" % (var, str(val)))
setattr(cls, var, val)
# Reset cached parameters since we have replaced configs already with ConfigValue
# so they are not parameters anymore to be re-evaluated when we do _get_parameters
if _FlowState.CACHED_PARAMETERS in cls._flow_state:
del cls._flow_state[_FlowState.CACHED_PARAMETERS]
# Run all the decorators. Step decorators are directly in the step and
# we will run those first and *then* we run all the flow level decorators
for step in cls._steps:
for deco in step.config_decorators:
if isinstance(deco, CustomStepDecorator):
debug.userconf_exec(
"Evaluating step level decorator %s for %s"
% (deco.__class__.__name__, step.name)
)
deco.evaluate(MutableStep(cls, step))
else:
raise MetaflowInternalError(
"A non CustomFlowDecorator found in step custom decorators"
)
if step.config_decorators:
# We remove all mention of the custom step decorator
setattr(cls, step.name, step)
mutable_flow = MutableFlow(cls)
for deco in cls._flow_state.get(_FlowState.CONFIG_DECORATORS, []):
if isinstance(deco, CustomFlowDecorator):
# Sanity check to make sure we are applying the decorator to the right
# class
if not deco._flow_cls == cls and not issubclass(cls, deco._flow_cls):
raise MetaflowInternalError(
"CustomFlowDecorator registered on the wrong flow -- "
"expected %s but got %s"
% (deco._flow_cls.__name__, cls.__name__)
)
debug.userconf_exec(
"Evaluating flow level decorator %s" % deco.__class__.__name__
)
deco.evaluate(mutable_flow)
# We reset cached_parameters on the very off chance that the user added
# more configurations based on the configuration
if _FlowState.CACHED_PARAMETERS in cls._flow_state:
del cls._flow_state[_FlowState.CACHED_PARAMETERS]
else:
raise MetaflowInternalError(
"A non CustomFlowDecorator found in flow custom decorators"
)
# Process parameters to allow them to also use config values easily
for var, param in cls._get_parameters():
if param.IS_CONFIG_PARAMETER:
continue
param.init()
# Reset all configs that were already present in the class.
# TODO: This means that users can't override configs directly. Not sure if this
# is a pattern we want to support
for var, param in to_reset_configs:
setattr(cls, var, param)
# Reset cached parameters again since we added back the config parameters
if _FlowState.CACHED_PARAMETERS in cls._flow_state:
del cls._flow_state[_FlowState.CACHED_PARAMETERS]
# Set the current flow class we are in (the one we just created)
parameters.replace_flow_context(cls)
# Re-calculate class level attributes after modifying the class
cls._init_attrs()
return cls