import datetime

import prefect

@prefect.task(max_retries=10, retry_delay=datetime.timedelta(seconds=900))

def task_a():

if prefect.context.task_run_count < 2:

raise Exception()

return 'a result'

@prefect.task

def task_b(a, c):

return 'b result'

with prefect.Flow('flow name') as flow:

a = task_a()

b = task_b.map(prefect.unmapped(a), [1, 2, 3])

flow.run_config = prefect.run_configs.DockerRun()

flow.storage = prefect.storage.Docker(

# python image

)

flow.register('project')