Hi again :smile: Im renamed a flow run from a task...
# prefect-server
r
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
Hey, @Robert Hales, do you have a small code snippet of what you’re trying to do?
r
Sure! I have a task like this
Copy code
@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
And then where do you use the
create_flow_run
? In a downstream task?
r
But when this task runs in the flow
Copy code
flow_run_id = create_flow_run(
        flow_name="example",
        upstream_tasks=[rename],
    )
the sub flow run is named incorrectly
k
Ok and the
create_flow_run
is the
StartFlowRun
task? or is it
client.create_flow_run()
?
r
create_flow_run
is the prefect provided task so the flow looks like this
Copy code
with Flow("flow") as flow:
    rename_task = rename(1234)
    
    flow_run_id = create_flow_run(
            flow_name="example",
            upstream_tasks=[rename_task],
        )
k
Ok I’ll try to replicate this
r
Thanks!
k
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:
Copy code
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
in
create_flow_run
there is this code
Copy 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
Ah I fully understand now. Just to be sure, is
dizzy-panda
the original name of Flow A in this case?
r
Exactly
Im imagining the issue is in this line
prefect.context.get("flow_run_name")
and the context having not updated
k
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
hmmm thats a shame, so context is set once a flow run starts?
k
Yes it is
r
Is that mentioned in the docs any where, i couldnt see any mention?
k
I don’t think it’s mentioned in the docs. You can verify this with a flow like this:
Copy code
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
If the context is set at flow run start how come it can contain the current task name?
k
It’s slugified on registration and stored in the database.
r
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
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
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
Ah I see. No problem!