https://prefect.io logo
Join Slack
Powered by
Hello, is it possible to dynamically set which `pr...
# prefect-community
w
Hello, is it possible to dynamically set which 
project
a 
create_flow_run
call creates a Flow for depending on environment variables? Example in thread
project name is parameterized by environment we’d like to run this Flow in
Copy code
with Flow("example flow") as flow:
    load_task = ...
    run_reporting_flow = create_flow_run(
        flow_name="Example Reporting Flow Name",
        project_name=f"reporting-{os.getenv(ENVIRONMENT)}",
        task_args=dict(trigger=all_successful),
    )
    run_reporting_flow.set_upstream(load_task)
after pushing code up to Github (our Flow storage mechanism) & registering it successfully using CI w/ the appropriate env vars set, running into:
Copy code
KeyError: 'Task slug create_flow_run-1 is not found in the current Flow. This is usually caused by a mismatch between the flow version stored in the Prefect backend and the flow that was loaded from storage.\n- Did you change the flow without re-registering it?\n- Did you register the flow without updating it in your storage location (if applicable)?'
k
You can if you wrap that f-string as a task. The problem here is that the f-string is evaluated as the DAG is constructed, not during flow run time.
w
gotcha, i’ll try that. when exactly does DAG construction happen? still confused why there’s a mismatch since the 
ENVIRONMMENT
in CI registration & the execution environments are the same
k
with Flow...
constructs and than you do either 
flow.run()
or 
flow.register()
but if you don’t call it, that’s why nothing runs. It just gets built
👍 1
w
thanks for the context Kevin! the error I was running into before was actually on the CI side - where the incorrect github storage 
ref
was being set (a branch that didn’t have the 
create_flow_run
call in the Flow was being referenced at Flow run time)
👍 1
k
Glad you figured it out
20 Views
Open in Slack
PreviousNext
Powered by