Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

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 <@U01HPF1476W>
Any `Task` class takes a `max_retries` argument, if you pass `max_retries` to `super().init` , your custom task will use this value

```class CustomTask(Task):
  def __init__(self, my_arg, **kwargs):
    self.prop = my_arg
    super().__init__(**kwargs)```

```my_custom_task = CustomTask(my_arg='foo', kwargs={'max_retries':100})```

<@U01SRTRJC0Y> 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 :sweat_smile:

<@UMMTFT29X> yep! I got my example from there