Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if path:
name = path[0]
if name[0].isdigit():
name = 's' + name
if hasattr(self, name):
getattr(self, name).add(func, path[1:])
else:
setattr(self, name, FunctionWrapper(func, path[1:]))
else:
self._func = func
def __call__(self, *args, **kwargs):
return self._func(*args, **kwargs)
class NestedState(State):
""" A state which allows substates.
Attributes:
parent (NestedState): The parent of the current state.
children (list): A list of child states of the current state.
"""
separator = '_'
u""" Separator between the names of parent and child states. In case '_' is required for
naming state, this value can be set to other values such as '.' or even unicode characters
such as '↦' (limited to Python 3 though).
"""
def __init__(self, name, on_enter=None, on_exit=None, ignore_invalid_triggers=None, parent=None, initial=None):
if parent is not None and isinstance(name, Enum):
raise AttributeError("NestedState does not support nested enumerations.")
import logging
import asyncio
import contextvars
from functools import partial
from ..core import State, Condition, Transition, EventData
from ..core import Event, MachineError, Machine
_LOGGER = logging.getLogger(__name__)
_LOGGER.addHandler(logging.NullHandler())
is_subtask = contextvars.ContextVar('is_subtask', default=False)
class AsyncState(State):
"""A persistent representation of a state managed by a ``Machine``. Callback execution is done asynchronously.
Attributes:
name (str): State name which is also assigned to the model(s).
on_enter (list): Callbacks awaited when a state is entered.
on_exit (list): Callbacks awaited when a state is entered.
ignore_invalid_triggers (bool): Indicates if unhandled/invalid triggers should raise an exception.
"""
async def enter(self, event_data):
""" Triggered when a state is entered. """
_LOGGER.debug("%sEntering state %s. Processing callbacks...", event_data.machine.name, self.name)
await event_data.machine.callbacks(self.on_enter, event_data)
_LOGGER.info("%sEntered state %s", event_data.machine.name, self.name)
async def exit(self, event_data):
accepted = kwargs.pop('accepted', False)
if accepted:
tags.append('accepted')
kwargs['tags'] = tags
super(Error, self).__init__(*args, **kwargs)
def enter(self, event_data):
""" Extends transitions.core.State.enter. Throws a `MachineError` if there is
no leaving transition from this state and 'accepted' is not in self.tags.
"""
if not event_data.machine.get_triggers(self.name) and not self.is_accepted:
raise MachineError("Error state '{0}' reached!".format(self.name))
super(Error, self).enter(event_data)
class Timeout(State):
""" Adds timeout functionality to a state. Timeouts are handled model-specific.
Attributes:
timeout (float): Seconds after which a timeout function should be called.
on_timeout (list): Functions to call when a timeout is triggered.
"""
dynamic_methods = ['on_timeout']
def __init__(self, *args, **kwargs):
"""
Args:
**kwargs: If kwargs contain 'timeout', assign the float value to self.timeout. If timeout
is set, 'on_timeout' needs to be passed with kwargs as well or an AttributeError will
be thrown. If timeout is not passed or equal 0.
"""
self.timeout = kwargs.pop('timeout', 0)
-----------------------------
This module contains mix ins which can be used to extend state functionality.
"""
from threading import Timer
import logging
import inspect
from ..core import MachineError, listify, State
_LOGGER = logging.getLogger(__name__)
_LOGGER.addHandler(logging.NullHandler())
class Tags(State):
""" Allows states to be tagged.
Attributes:
tags (list): A list of tag strings. `State.is_` may be used
to check if is in the list.
"""
def __init__(self, *args, **kwargs):
"""
Args:
**kwargs: If kwargs contains `tags`, assign them to the attribute.
"""
self.tags = kwargs.pop('tags', [])
super(Tags, self).__init__(*args, **kwargs)
def __getattr__(self, item):
if item.startswith('is_'):
return item[3:] in self.tags
for callback in self.on_timeout:
event_data.machine.callback(callback, event_data)
_LOGGER.info("%sTimeout state %s processed.", event_data.machine.name, self.name)
@property
def on_timeout(self):
""" List of strings and callables to be called when the state timeouts. """
return self._on_timeout
@on_timeout.setter
def on_timeout(self, value):
""" Listifies passed values and assigns them to on_timeout."""
self._on_timeout = listify(value)
class Volatile(State):
""" Adds scopes/temporal variables to the otherwise persistent state objects.
Attributes:
volatile_cls (cls): Class of the temporal object to be initiated.
volatile_hook (str): Model attribute name which will contain the volatile instance.
"""
def __init__(self, *args, **kwargs):
"""
Args:
**kwargs: If kwargs contains `volatile`, always create an instance of the passed class
whenever the state is entered. The instance is assigned to a model attribute which
can be passed with the kwargs keyword `hook`. If hook is not passed, the instance will
be assigned to the 'attribute' scope. If `volatile` is not passed, an empty object will
be assigned to the model's hook.
"""
self.volatile_cls = kwargs.pop('volatile', VolatileObject)
def _has_state(self, state):
if isinstance(state, State):
if state in self.states.values():
return True
else:
raise ValueError('State %s has not been added to the machine' % state.name)
else:
return False
queued (bool): Whether transitions in callbacks should be executed immediately (False) or sequentially.
send_event (bool): When True, any arguments passed to trigger methods will be wrapped in an EventData
object, allowing indirect and encapsulated access to data. When False, all positional and keyword
arguments will be passed directly to all callback methods.
auto_transitions (bool): When True (default), every state will automatically have an associated
to_{state}() convenience trigger in the base model.
ignore_invalid_triggers (bool): When True, any calls to trigger methods that are not valid for the
present state (e.g., calling an a_to_b() trigger when the current state is c) will be silently
ignored rather than raising an invalid transition exception.
name (str): Name of the ``Machine`` instance mainly used for easier log message distinction.
"""
separator = '_' # separates callback type from state/transition name
wildcard_all = '*' # will be expanded to ALL states
wildcard_same = '=' # will be expanded to source state
state_cls = State
transition_cls = Transition
event_cls = Event
def __init__(self, model='self', states=None, initial='initial', transitions=None,
send_event=False, auto_transitions=True,
ordered_transitions=False, ignore_invalid_triggers=None,
before_state_change=None, after_state_change=None, name=None,
queued=False, prepare_event=None, finalize_event=None, **kwargs):
"""
Args:
model (object or list): The object(s) whose states we want to manage. If 'self',
the current Machine instance will be used the model (i.e., all
triggering events will be attached to the Machine itself). Note that an empty list
is treated like no model.
states (list or Enum): A list or enumeration of valid states. Each list element can be either a
string, an enum member or a State instance. If string or enum member, a new generic State
def update(self, state):
""" Updates the EventData object with the passed state.
Attributes:
state (State, str or Enum): The state object, enum member or string to assign to EventData.
"""
if not isinstance(state, State):
self.state = self.machine.get_state(state)