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}")
if self.indirect:
indirect = _indirect(self.node if self.indirect is True else self.indirect)
has_indirect = has_node(config, indirect)
else:
has_indirect = False
node = get_node(config, indirect) if has_indirect else self.node
if node is None:
logger.debug(f" bind param '{param.name}' to the whole state")
else:
logger.debug(f" bind param '{param.name}' to state node '{node}'")
binding = Pipe.Node.Binding()
binding.node = node
binding.root = state
binding.root_name = "state"
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 not None and is_mutable(value) and not self.mutable:
raise AttributeError(f"param '{param.name}' is mutable but not marked as such")
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 not self.mutable:
raise AttributeError(f"param '{param.name}' is not mutable")
if binding.node != node or binding.root is not state or binding.root_name != "state":
binding.node = node
binding.root = state
binding.root_name = "state"
logger.debug(f" re-bind param '{param.name}' to {binding.root_name} node '{binding.node}'")
set_node(binding.root, binding.node, value)
return binding, getter, setter