Skip to content

trestle.core.models.plans

trestle.core.models.plans ¤

Plan of action of a command.

Attributes¤

logger = logging.getLogger(__name__) module-attribute ¤

Classes¤

Plan ¤

Plan of action of a command.

Source code in trestle/core/models/plans.py
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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

        # Compare actions by their content, not by object identity
        if len(self.get_actions()) != len(other.get_actions()):
            return False

        for i, (self_action, other_action) in enumerate(zip(self.get_actions(), other.get_actions(), strict=True)):
            if self_action != other_action:
                # Debug: print which action differs
                logger.debug(f'Action {i} differs:')
                logger.debug(f'  Self: {self_action}')
                logger.debug(f'  Other: {other_action}')
                return False

        return True

    def __repr__(self) -> str:
        """Return a detailed representation of the plan."""
        return f'Plan(actions={[str(a) for a in self._actions]})'
Methods:¤
__eq__(other) ¤

Check that two plans are equal.

Source code in trestle/core/models/plans.py
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def __eq__(self, other: object) -> bool:
    """Check that two plans are equal."""
    if not isinstance(other, Plan):
        return False

    # Compare actions by their content, not by object identity
    if len(self.get_actions()) != len(other.get_actions()):
        return False

    for i, (self_action, other_action) in enumerate(zip(self.get_actions(), other.get_actions(), strict=True)):
        if self_action != other_action:
            # Debug: print which action differs
            logger.debug(f'Action {i} differs:')
            logger.debug(f'  Self: {self_action}')
            logger.debug(f'  Other: {other_action}')
            return False

    return True
__init__() ¤

Initialize a plan.

Source code in trestle/core/models/plans.py
28
29
30
def __init__(self) -> None:
    """Initialize a plan."""
    self._actions: List[Action] = []
__repr__() ¤

Return a detailed representation of the plan.

Source code in trestle/core/models/plans.py
 99
100
101
def __repr__(self) -> str:
    """Return a detailed representation of the plan."""
    return f'Plan(actions={[str(a) for a in self._actions]})'
__str__() ¤

Print the plan.

Source code in trestle/core/models/plans.py
35
36
37
38
39
40
41
42
43
44
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(action) ¤

Add a new action.

Source code in trestle/core/models/plans.py
50
51
52
def add_action(self, action: Action) -> None:
    """Add a new action."""
    self._actions.append(action)
add_actions(actions) ¤

Add actions in order.

Source code in trestle/core/models/plans.py
54
55
56
def add_actions(self, actions: List[Action]) -> None:
    """Add actions in order."""
    self._actions.extend(actions)
clear_actions() ¤

Clear all actions.

Source code in trestle/core/models/plans.py
58
59
60
def clear_actions(self) -> None:
    """Clear all actions."""
    self._actions = []
execute() ¤

Execute the actions in the plan.

Source code in trestle/core/models/plans.py
62
63
64
65
66
67
68
69
70
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() ¤

Get all actions.

Source code in trestle/core/models/plans.py
46
47
48
def get_actions(self) -> List[Action]:
    """Get all actions."""
    return self._actions
rollback() ¤

Rollback the actions in the plan.

Source code in trestle/core/models/plans.py
72
73
74
75
76
77
78
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