k

    Kevin Weiler

    11 months ago
    Hi there! I’d like to use the
    RenameFlowRun
    prefect task to rename my flow based on a value in the context. I noticed that if I try something like:
    rename_flow = RenameFlowRun(flow_name=f"{prefect.context.today}")
    It tries to evaluate the context at deploy time, instead of at runtime. Is there a reasonable way to do this? @Rob Douglas ended up just pulling the guts out of the
    run()
    method of
    RenameFlowRun
    and putting it in a task. That works, but it seems a bit dodgy. PS - the documentation for this is wrong, as the example shows importing
    FlowRenameTask
    instead of
    RenameFlowRun
    Kevin Kho

    Kevin Kho

    11 months ago
    Could you think that example? So stuff in the Flow block and these
    init
    are executed immediately so even if you did something like
    datetime.datetime.now()
    it would be persisted during build. The way around this is to create a function that returns
    prefect.context.today
    so that it executes during runtime. And then I think you want,
    with Flow(...) as flow:
         RenameFlowRun()(flow_name=functioncall())
    The first parenthesis is the init. The second in the run.
    k

    Kevin Weiler

    11 months ago
    that’s interesting - so this:
    with Flow(...) as flow:
         RenameFlowRun()(flow_name=prefect.context.today)
    Wouldn’t work because the arguments are evaluated at deploy time correct?
    Kevin Kho

    Kevin Kho

    11 months ago
    Yes that’s right, but really to can just wrap it in a lambda too
    with Flow(...) as flow:
         RenameFlowRun()(flow_name=task(lambda **args: prefect.context.today))
    i think this will work
    k

    Kevin Weiler

    11 months ago
    yep - that makes sense - will give that a try - thanks!
    I guess this will still yield 2 tasks in the graph, one to return the context value, and one to rename the flow
    Kevin Kho

    Kevin Kho

    11 months ago
    That will yeah. If you want it to be 1, I think maybe try using a function instead of a task and see if it works. If not, you probably need to modify the task, but modifying the tasks in our task library is not bad practice btw. A lot of people do it. Specifically, you can subclass it and add a few lines then call
    super.run()
    .