Robert Hales
07/15/2021, 9:55 AMclient.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?Kevin Kho
07/15/2021, 1:55 PMRobert Hales
07/15/2021, 2:00 PM@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 UIKevin Kho
07/15/2021, 2:01 PMcreate_flow_run
? In a downstream task?Robert Hales
07/15/2021, 2:01 PMflow_run_id = create_flow_run(
flow_name="example",
upstream_tasks=[rename],
)
Kevin Kho
07/15/2021, 2:03 PMcreate_flow_run
is the StartFlowRun
task? or is it client.create_flow_run()
?Robert Hales
07/15/2021, 2:05 PMcreate_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],
)
Kevin Kho
07/15/2021, 2:08 PMRobert Hales
07/15/2021, 2:08 PMKevin Kho
07/15/2021, 3:13 PMStartFlowRun
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?Robert Hales
07/15/2021, 3:15 PMcreate_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}"
Parent
lets sayParent-Flow B
dizzy-panda-Flow B
for exampleKevin Kho
07/15/2021, 3:19 PMdizzy-panda
the original name of Flow A in this case?Robert Hales
07/15/2021, 3:20 PMprefect.context.get("flow_run_name")
and the context having not updatedKevin Kho
07/15/2021, 3:23 PMrun_name
to the StartFlowRun
so that this if
statement is not reached.Robert Hales
07/15/2021, 3:26 PMKevin Kho
07/15/2021, 3:26 PMRobert Hales
07/15/2021, 3:27 PMKevin Kho
07/15/2021, 3:40 PMfrom 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")
context-test
. Second prints test
. But the last prints context-test
.Robert Hales
07/15/2021, 3:44 PMKevin Kho
07/15/2021, 3:46 PMRobert Hales
07/15/2021, 3:57 PMclient.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 updatesKevin Kho
07/15/2021, 4:00 PMrun_name
to StartFlowRun not work for you use case?Robert Hales
07/15/2021, 4:03 PMKevin Kho
07/15/2021, 4:05 PM