def trigger_state_machine()

in assets/functions/state_topic_subscription/app.py [0:0]

• 14 lines of code
• 2 McCabe index (conditional complexity)

def trigger_state_machine():
    state_machine_arn = os.environ['ml_pipeline_trigger_state_machine']

    running_executions = client.list_executions(
        stateMachineArn=state_machine_arn,
        statusFilter='RUNNING',
        maxResults=10
    )

    if not running_executions['executions']:
        print('start new execution')
        boto3 \
            .client('stepfunctions') \
            .start_execution(stateMachineArn=state_machine_arn)
    else:
        print('state machine already running.')