plans
trestle.core.models.plans
¤
Plan of action of a command.
logger
¤
Classes¤
Plan
¤
Plan of action of a command.
Source code in trestle/core/models/plans.py
class Plan:
"""Plan of action of a command."""
def __init__(self) -> None:
"""Initialize a plan."""
self._actions: List[Action] = []
def _action_key(self, action: Action) -> int:
return hash(action)
def __str__(self) -> str:
"""Print the plan."""
list_actions = []
index = 1
for action in self._actions:
list_actions.append(f'{index}. {action}')
index = index + 1
list_str = '\n'.join(list_actions)
return list_str
def get_actions(self) -> List[Action]:
"""Get all actions."""
return self._actions
def add_action(self, action: Action) -> None:
"""Add a new action."""
self._actions.append(action)
def add_actions(self, actions: List[Action]) -> None:
"""Add actions in order."""
self._actions.extend(actions)
def clear_actions(self) -> None:
"""Clear all actions."""
self._actions = []
def execute(self) -> None:
"""Execute the actions in the plan."""
for action in self._actions:
try:
action.execute()
except Exception as e:
logger.error(f'Failed to execute action {action} for the plan: {e}. Rolling back.')
self.rollback()
raise e
def rollback(self) -> None:
"""Rollback the actions in the plan."""
# execute in reverse order
for action in reversed(self._actions):
if action.has_rollback() is False:
raise UnsupportedOperation(f'{action.get_type()} does not support rollback')
action.rollback()
def __eq__(self, other: object) -> bool:
"""Check that two plans are equal."""
if not isinstance(other, Plan):
return False
return self.get_actions() == other.get_actions()
Methods¤
__eq__(self, other)
special
¤
Check that two plans are equal.
Source code in trestle/core/models/plans.py
def __eq__(self, other: object) -> bool:
"""Check that two plans are equal."""
if not isinstance(other, Plan):
return False
return self.get_actions() == other.get_actions()
__init__(self)
special
¤
Initialize a plan.
Source code in trestle/core/models/plans.py
def __init__(self) -> None:
"""Initialize a plan."""
self._actions: List[Action] = []
__str__(self)
special
¤
Print the plan.
Source code in trestle/core/models/plans.py
def __str__(self) -> str:
"""Print the plan."""
list_actions = []
index = 1
for action in self._actions:
list_actions.append(f'{index}. {action}')
index = index + 1
list_str = '\n'.join(list_actions)
return list_str
add_action(self, action)
¤
Add a new action.
Source code in trestle/core/models/plans.py
def add_action(self, action: Action) -> None:
"""Add a new action."""
self._actions.append(action)
add_actions(self, actions)
¤
Add actions in order.
Source code in trestle/core/models/plans.py
def add_actions(self, actions: List[Action]) -> None:
"""Add actions in order."""
self._actions.extend(actions)
clear_actions(self)
¤
Clear all actions.
Source code in trestle/core/models/plans.py
def clear_actions(self) -> None:
"""Clear all actions."""
self._actions = []
execute(self)
¤
Execute the actions in the plan.
Source code in trestle/core/models/plans.py
def execute(self) -> None:
"""Execute the actions in the plan."""
for action in self._actions:
try:
action.execute()
except Exception as e:
logger.error(f'Failed to execute action {action} for the plan: {e}. Rolling back.')
self.rollback()
raise e
get_actions(self)
¤
Get all actions.
Source code in trestle/core/models/plans.py
def get_actions(self) -> List[Action]:
"""Get all actions."""
return self._actions
rollback(self)
¤
Rollback the actions in the plan.
Source code in trestle/core/models/plans.py
def rollback(self) -> None:
"""Rollback the actions in the plan."""
# execute in reverse order
for action in reversed(self._actions):
if action.has_rollback() is False:
raise UnsupportedOperation(f'{action.get_type()} does not support rollback')
action.rollback()
handler: python