Hello all, I see that I can set up retry logic with

@task(max_retries=3)

but how can I set up the retry logic if I am not setting up tasks with the functional API? I do not see anything in the docs that explains. Thank you in advance!

Hi @Matthew Blau Any

Task

class takes a

max_retries

argument, if you pass

max_retries

to

super().init

, your custom task will use this value

@Zach Angell so in my case I have a flow that looks like

with Flow(name="integration",          state_handlers=[slack_notifier],) as flow:    start_container = start(container_id=container)    code = status_code(container_id=container, upstream_tasks=[start_container]) collect_logs = logs(container_id=container, upstream_tasks=[code])

so what you're saying is that I can pass in {'max_retires':3} to the start_container task?

I think Zach’s approach is when creating your Task by inheriting the Task class. Can I see the definition of the

start_container

task you have?

start = StartContainer(max_retries=3)
with Flow(name="integration", state_handlers=[slack_notifier],) as flow:
    start_container = start(container_id=container)    
    code = status_code(container_id=container, upstream_tasks=[start_container]) 
    collect_logs = logs(container_id=container, upstream_tasks=[code])

tasks need to be first initialized, and then called. Assuming you got your example from: https://docs.prefect.io/core/examples/functional_docker.html btw 😅

I see. Thanks for that @emre!

@emre yep! I got my example from there