Hi everyone, I have a common set of concerns (configuration, logging) I want to setup for each task and my first idea was to override the .run() method of the prefect.tasks.core.function.FunctionTask and use prefect @task decorator with the new FunctionTask, but this method is not being called in the runtime when executing a task? Is it possible to add some cross cutting concern code before running a task itself? Tried a decorator, but seems it is not pickable by cloudpickle. I'll add a code example in the comment

Hey @domagoj jercic, maybe this message will show you how to make a decorator

Hi Kevin, helps but still doesn't work when I set flow executor to dask executor

flow.executor = LocalDaskExecutor(scheduler="multiprocessing")

Is there a way to make the decorator pickable?

Ok I’ll have to try more stuff on my end. I’ll get back to you later today probably

Thank you. I'm UTC+1 so I'll probably respond tomorrow 😄

Hey I didn’t get a chance yesterday. Will try today

thank you. Let me know if I can help somehow

Hey it seems to work for me with this setup:

import traceback
from prefect import task, Flow
from functools import partial, wraps
import prefect
from prefect.executors import LocalDaskExecutor

def custom_task(func=None, **task_init_kwargs):
    if func is None:
        return partial(custom_task, **task_init_kwargs)

    @wraps(func)
    def safe_func(**kwargs):
        try:
            return func(**kwargs)
        except Exception as e:
            print(f"Full Traceback: {traceback.format_exc()}")
            raise RuntimeError(type(e)) from None  # from None is necessary to not log the stacktrace

    safe_func.__name__ = func.__name__
    return task(safe_func, **task_init_kwargs)

@custom_task
def abc(x):
    return x

with Flow("custom-decorator-test") as flow:
    abc.map([1,2,3,4,5])

flow.executor = LocalDaskExecutor(scheduler="processes")

Hm, seems like it. The same code works when I put it as a standalone file. Does that mean that part of the flow is not pickable?