How to use the stepfunctions.exceptions.DuplicateStatesInChain function in stepfunctions

To help you get started, we’ve selected a few stepfunctions examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github aws / aws-step-functions-data-science-sdk-python / tests / unit / test_steps.py View on Github external
def test_chaining_steps():
    s1 = Pass('Step - One')
    s2 = Pass('Step - Two')
    s3 = Pass('Step - Three')

    Chain([s1, s2])
    assert s1.next_step == s2
    assert s2.next_step is None

    chain1 = Chain([s2, s3])
    assert s2.next_step == s3

    chain2 = Chain([s1, s3])
    assert s1.next_step == s3
    assert s2.next_step == s1.next_step
    with pytest.raises(DuplicateStatesInChain):
        chain2.append(s3)

    with pytest.raises(DuplicateStatesInChain):
        chain3 = Chain([chain1, chain2])
    
    s1.next(s2)
    chain3 = Chain([s3, s1])
    assert chain3.steps == [s3, s1]
    assert s3.next_step == s1
    assert s1.next_step == s2
    assert s2.next_step == s3

    Chain([Chain([s3]), Chain([s1])])

    with pytest.raises(DuplicateStatesInChain):
        Chain([Chain([s1, s2, s1]), s3])
github aws / aws-step-functions-data-science-sdk-python / tests / unit / test_steps.py View on Github external
s3 = Pass('Step - Three')

    Chain([s1, s2])
    assert s1.next_step == s2
    assert s2.next_step is None

    chain1 = Chain([s2, s3])
    assert s2.next_step == s3

    chain2 = Chain([s1, s3])
    assert s1.next_step == s3
    assert s2.next_step == s1.next_step
    with pytest.raises(DuplicateStatesInChain):
        chain2.append(s3)

    with pytest.raises(DuplicateStatesInChain):
        chain3 = Chain([chain1, chain2])
    
    s1.next(s2)
    chain3 = Chain([s3, s1])
    assert chain3.steps == [s3, s1]
    assert s3.next_step == s1
    assert s1.next_step == s2
    assert s2.next_step == s3

    Chain([Chain([s3]), Chain([s1])])

    with pytest.raises(DuplicateStatesInChain):
        Chain([Chain([s1, s2, s1]), s3])
        Chain([s1, s2, s1, s3])
    Chain([Chain([s1, s2]), s3])
    assert s1.next_step == s2
github aws / aws-step-functions-data-science-sdk-python / src / stepfunctions / steps / states.py View on Github external
def append(self, step):
        """Add a state at the tail end of the chain.

        Args:
            step (State): State to insert at the tail end of the chain.
        """
        if len(self.steps) == 0:
            self.steps.append(step)
        else:
            if step in self.steps:
                raise DuplicateStatesInChain("State '{step_name}' is already inside this chain. A chain cannot have duplicate states.".format(step_name=step.state_id))
            last_step = self.steps[-1]
            last_step.next(step)
            self.steps.append(step)
github aws / aws-step-functions-data-science-sdk-python / src / stepfunctions / steps / states.py View on Github external
def __init__(self, steps=[]):
        """
        Args:
            steps (list(State), optional): List of states to be chained in-order. (default: [])
        """
        if not isinstance(steps, list):
            raise TypeError("Chain takes a 'list' of steps. You provided an input that is not a list.")
        self.steps = []

        steps_expanded = []
        [steps_expanded.extend(step) if isinstance(step, Chain) else steps_expanded.append(step) for step in steps]
        for step in steps_expanded:
            if steps_expanded.count(step) > 1:
                raise DuplicateStatesInChain("Duplicate states in the chain.")
        list(map(self.append, steps_expanded))