b

    Bastian Röhrig

    1 year ago
    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?
    Kevin Kho

    Kevin Kho

    1 year ago
    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:
    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

    Bastian Röhrig

    1 year ago
    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?
    Kevin Kho

    Kevin Kho

    1 year ago
    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.