File: state_wait.py

package info (click to toggle)
python-moto 5.1.18-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 116,520 kB
  • sloc: python: 636,725; javascript: 181; makefile: 39; sh: 3
file content (33 lines) | stat: -rw-r--r-- 1,197 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from __future__ import annotations

from moto.stepfunctions.parser.api import HistoryEventType
from moto.stepfunctions.parser.asl.component.state.state import CommonStateField
from moto.stepfunctions.parser.asl.component.state.state_props import StateProps
from moto.stepfunctions.parser.asl.component.state.wait.wait_function.wait_function import (
    WaitFunction,
)
from moto.stepfunctions.parser.asl.eval.environment import Environment


class StateWait(CommonStateField):
    wait_function: WaitFunction

    def __init__(self):
        super().__init__(
            state_entered_event_type=HistoryEventType.WaitStateEntered,
            state_exited_event_type=HistoryEventType.WaitStateExited,
        )

    def from_state_props(self, state_props: StateProps) -> None:
        super().from_state_props(state_props)
        self.wait_function = state_props.get(
            typ=WaitFunction,
            raise_on_missing=ValueError(
                f"Undefined WaitFunction for StateWait: '{self}'."
            ),
        )

    def _eval_state(self, env: Environment) -> None:
        self.wait_function.eval(env)
        if self.assign_decl:
            self.assign_decl.eval(env=env)