https://prefect.io logo
Title
r

Robert Hales

07/15/2021, 9:55 AM
Hi again 😄 Im renamed a flow run from a task using the
client.set_flow_run_name
method, and it sucessfully updates the UI. However, when I then do
create_flow_run
the subtasks use the old auto-generated name. Im assuming because the context is stale?
k

Kevin Kho

07/15/2021, 1:55 PM
Hey, @Robert Hales, do you have a small code snippet of what you’re trying to do?
r

Robert Hales

07/15/2021, 2:00 PM
Sure! I have a task like this
@task
def rename(id):
    obj = get_object_from_db(id)
    client = Client()
    client.set_flow_run_name(prefect.context.flow_run_id, obj.name)
Which runs before all the tasks in my flow and renames the flow run in the UI
k

Kevin Kho

07/15/2021, 2:01 PM
And then where do you use the
create_flow_run
? In a downstream task?
r

Robert Hales

07/15/2021, 2:01 PM
But when this task runs in the flow
flow_run_id = create_flow_run(
        flow_name="example",
        upstream_tasks=[rename],
    )
the sub flow run is named incorrectly
k

Kevin Kho

07/15/2021, 2:03 PM
Ok and the
create_flow_run
is the
StartFlowRun
task? or is it
client.create_flow_run()
?
r

Robert Hales

07/15/2021, 2:05 PM
create_flow_run
is the prefect provided task so the flow looks like this
with Flow("flow") as flow:
    rename_task = rename(1234)
    
    flow_run_id = create_flow_run(
            flow_name="example",
            upstream_tasks=[rename_task],
        )
k

Kevin Kho

07/15/2021, 2:08 PM
Ok I’ll try to replicate this
r

Robert Hales

07/15/2021, 2:08 PM
Thanks!
k

Kevin Kho

07/15/2021, 3:13 PM
Hey @Robert Hales, I think I’m not getting something here. You are running flow A, and then renaming the flow run, and then you start another flow B that you expect to have the same name as the renamed A? I think their flow runs are separate so they would have different names. If you want them to have the same name, I think you can use the
StartFlowRun
task like this:
StartFlowRun(flow_name=xxx, project_name=xxx, run_name=xxx, wait=True)
and then pass the same name into the run_name if that’s what you’re looking for. I also added the wait=True, cuz I think that’s what you wanted? This will be something like:
with Flow("flow") as flow:
    rename_task = rename(THE_RUN_NAME)
    flow_run_id = StartFlowRun(flow_name=xxx, project_name=xxx, run_name=THE_RUN_NAME, wait=True)(upstream_tasks=[])
Am I answering the question?
r

Robert Hales

07/15/2021, 3:15 PM
in
create_flow_run
there is this code
# Generate a 'sub-flow' run name
if not run_name:
    current_run = prefect.context.get("flow_run_name")
    if current_run:
        run_name = f"{current_run}-{flow.name}"
So I am renaming the parent flow run to be
Parent
lets say
I would expect the sub-flows to be called
Parent-Flow B
instead they are
dizzy-panda-Flow B
for example
k

Kevin Kho

07/15/2021, 3:19 PM
Ah I fully understand now. Just to be sure, is
dizzy-panda
the original name of Flow A in this case?
r

Robert Hales

07/15/2021, 3:20 PM
Exactly
Im imagining the issue is in this line
prefect.context.get("flow_run_name")
and the context having not updated
k

Kevin Kho

07/15/2021, 3:23 PM
Yeah you are probably right. Unfortunately, the context can’t be manipulated once it’s loaded in. I think in this case, you really have to pass the
run_name
to the
StartFlowRun
so that this
if
statement is not reached.
r

Robert Hales

07/15/2021, 3:26 PM
hmmm thats a shame, so context is set once a flow run starts?
k

Kevin Kho

07/15/2021, 3:26 PM
Yes it is
r

Robert Hales

07/15/2021, 3:27 PM
Is that mentioned in the docs any where, i couldnt see any mention?
k

Kevin Kho

07/15/2021, 3:40 PM
I don’t think it’s mentioned in the docs. You can verify this with a flow like this:
from prefect import Flow, Parameter, task
import prefect
from prefect.client.client import Client

@task
def abc(x):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(prefect.context.flow_name)
    prefect.context.flow_name = "test"
    <http://logger.info|logger.info>(prefect.context.flow_name)
    return x

@task
def bcd(x):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(prefect.context.flow_name)
    return x

with Flow("context-test") as flow:
    a = abc(1)
    b = bcd(a)

flow.register("testing")
First logger prints
context-test
. Second prints
test
. But the last prints
context-test
.
r

Robert Hales

07/15/2021, 3:44 PM
If the context is set at flow run start how come it can contain the current task name?
k

Kevin Kho

07/15/2021, 3:46 PM
It’s slugified on registration and stored in the database.
r

Robert Hales

07/15/2021, 3:57 PM
Does
client.set_flow_run_name(prefect.context.flow_run_id, obj.name)
not effect the database? it performas a graphql query, right? and the UI correctly updates
k

Kevin Kho

07/15/2021, 4:00 PM
I think it does, but I think it might not update the current context in the flow run. Why does passing the
run_name
to StartFlowRun not work for you use case?
r

Robert Hales

07/15/2021, 4:03 PM
I think it will do, im guessing ill need another task to build the string but that isnt an issue really. Just trying to gain an understanding of how data flows through prefect, as intuitively I would have thought that the client call would update context
Thanks for all your help!
k

Kevin Kho

07/15/2021, 4:05 PM
Ah I see. No problem!