Unable to set retries or other arguments in non decorator task mode:

[JackDapid]

Does anybody have an example how to set task arguments with the non-task-decorator notation?

I have no issues using the max_retries or other task arguments when I use the task decorator:

@task(max_retries=10, retry_delay=datetime.timedelta(seconds=5))
def my_prefect_task():
...

but than if I try to do more complex prefect task classes:

class MyPrefectTask(Task(max_retries=10, retry_delay=datetime.timedelta(seconds=5))):
    def __init__(self, df=None, **kwargs):
...

I get some binding errors:

ValueError: Could not infer an active Flow context.
/usr/local/lib/python3.7/site-packages/prefect/core/task.py:416: ValueError

[joshmeek]

@JackDapid Try setting the task arguments at initialization like this:

class MyPrefectTask(Task):
    def __init__(self, df=None, **kwargs):
        ....

my_task = MyPrefectTask(max_retries=10, retry_delay=datetime.timedelta(seconds=5))

[JackDapid]

@joshmeek, thanks for the super fast reaction, yes that works, but is rather uncomfortable in the context I am using it in.

While in decorator @task(...) case I am setting the 'default' retries for a task and than create as many instances as I need with these settings, with your proposed solution I would have to set these arg every time (or at every location) instead of having them centrally with the definition of the task. But maybe I naively assumed that setting of the default args like in the @task(...) decorator notation would also work in the Task() class notation - I will have a look what the @task decorator actually does as I assumed it would turn a normal function into a Task() class

Or is there another way to set the default retries, name, ... arguments?
🤔

[JackDapid]

prefect/src/prefect/utilities/tasks.py

Line 318 in 1f43f5a

def task(

->

prefect/src/prefect/tasks/core/function.py

Line 27 in 6f1c8e1

class FunctionTask(prefect.Task):

[joshmeek]

I believe you want something like this:

class MyPrefectTask(Task):
    def __init__(self, df=None, max_retries=10, **kwargs):
        ....
        super().__init__(max_retries=max_retries, **kwargs)

my_task = MyPrefectTask(retry_delay=datetime.timedelta(seconds=5))  # using default
my_task = MyPrefectTask(max_retries=10, retry_delay=datetime.timedelta(seconds=5))  # not using default

[JackDapid]

THX @joshmeek yes, that was it 👍

[cicdw]

FYI: you can technically achieve this without classes as well:

@task(max_retries=10, retry_delay=datetime.timedelta(seconds=5))
def my_prefect_task():
    pass


# within a Flow context
my_prefect_task(task_args={max_retries=5}) # overrides `max_retries`

I think @joshmeek 's suggestion is cleaner / easier to reason about but wanted to point this out for reference

[JackDapid]

Thanks @cicdw, yes I have used the decorator notation @task for that successfully before, I use the class notation for other reasons, but I totally agree fro others that have the same questions your comment will be very helpfull, thank you