https://prefect.io
Join Slack
is there a way to set the project_name parameter of the StartFlowRun Task automatically to the curre...
# ask-community
x
is there a way to set the project_name parameter of the StartFlowRun Task automatically to the current project of the flow or None if its run with core flow.run?
Unfortunately the context doesn't seem to contain the project name
a
project_name
should be in the context if you're using prefect cloud or prefect server, but not prefect core.
x
thats interesting... I didn't find it listed on https://docs.prefect.io/api/latest/utilities/context.html#context-2
a
It was probably an oversight when the change was made. Refer to the changelog for version 0.14.11
Oh, but wait. You mentioned "core flow.run". Unfortunately no, in that case it isn't added to the context.
x
Actually I just want to get none for flow.run, so it should be fine
doesn't work somehow when triggering from server... ist context unavailable with the flow definition?
As long as I start it with server it works if i use a helper tasks:
Copy code
@task()
def get_project_name():
    return prefect.context.get("project_name")
however when I run it with flow.run I get the following error: 
ValueError: Must provide a project name.
Which is not what I expected as the docs say the following regarding project name:
Copy code
If running with Prefect Core's server as the backend, this should not be provided.
looking at the source, it seems to only consider the cloud/server scenario, not local run
for now I've disabled triggering the second flow on flow.run by using the following method:
Copy code
@task()
def get_project_meta() -> Tuple[str, bool]:
    project_name = prefect.context.get("project_name")
    has_project_name = bool(project_name)
    return project_name, has_project_name


flow_run = StartFlowRun()
with Flow("my-flow") as flow:
    ...
    project_name, has_project_name = get_project_meta()
    with case(has_project_name, True):
        flow_run(flow_name='my-flow-check', project_name=project_name)
If anyone had an idea on how to make this less verbose I'd appreciate it
z
Hey @xyzz -- you could write a wrapper for the 
StartFlowRun
task that does this for you e.g.
Copy code
@task
def start_flow_run_with_prj(flow_name):
    project_name = prefect.context.get("project_name")
    if not project_name:
         return
    StartFlowRun(flow_name=flow_name, project_name=project_name).run()
x
That's a great idea, thanks!
2 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by