in core/__init__.py [0:0]
def handle_param(self, param, config, state, logger):
if param.default is not param.empty and is_mutable(param.default):
raise TypeError(f"param '{param.name}': mutable default not allowed: {param.default}")
has_value = has_node(config, self.node)
has_indirect = has_node(config, _indirect(self.node))
if has_value and has_indirect:
raise ConfigError(f"param '{param.name}': config cannot specify both '{self.node}' and '{_indirect(self.node)}'")
binding = Pipe.Node.Binding()
if has_indirect:
binding.node = get_node(config, _indirect(self.node))
binding.root = state
binding.root_name = "state"
else:
binding.node = self.node
binding.root = config
binding.root_name = "config"
logger.debug(f" bind param '{param.name}' to {binding.root_name} node '{binding.node}'")
def default_action():
if param.default is param.empty:
raise KeyError(f"param '{param.name}': {binding.root_name} node not found: '{binding.node}'")
return param.default
def getter(_):
value = get_node(binding.root, binding.node, default_action=default_action)
if value is None or param.type is Any or isinstance(value, param.type):
return value
value_type = type(value).__name__
expected_type = param.type.__name__
raise Error(
f"param '{param.name}': {binding.root_name} node '{binding.node}' type mismatch: '{value_type}' (expected '{expected_type}')"
)
def setter(_, value):
if binding.node != self.node or binding.root is not config or binding.root_name != "config":
binding.node = self.node
binding.root = config
binding.root_name = "config"
logger.debug(f" re-bind param '{param.name}' to {binding.root_name} node '{binding.node}'")
config.pop(_indirect(self.node))
set_node(binding.root, binding.node, value)
return binding, getter, setter