Hey everyone, I am using the `create_flow_run` and...
# ask-community
b
Hey everyone, I am using the
create_flow_run
and
get_task_run_result
to interact with child flows. Now I would love to set some general task properties like
state_handlers
and
task_run_name
on these. Maybe I am missing something, but I can't figure out how to do this without writing my own wrapper task. Is there a way to do this?
k
Hey @Bastian Röhrig, you can’t set these from the
create_flow_run
call through parameters if that is what you’re saying. If you’re saying you have a common state handler that you want to attach to all tasks, then yes it’s very common for users to either use a function to set it, or to make their own version of the
task
decorator.
Sample here:
Copy code
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
b
Yeah, I wanted to set them from the
create_flow_run
call through parameters. Do you happen to know why that is not supported or are there plans to support it in the future?
k
Because the
state_handler
needs to be attached during registration. The task is already registered by the name you are doing
create_flow_run
. But there is a workaround here. You can do
prefect.context.get("parameters")
inside a
state_handler
. This returns a dict of values, and then you can do an
if-else
based on that. The
task_run_name
can be altered based on inputs so you shouldn’t have a problem there.