Thread
#prefect-community
    w

    Wesley Jin

    6 months ago
    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
    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:
    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)?'
    Kevin Kho

    Kevin Kho

    6 months ago
    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

    Wesley Jin

    6 months ago
    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
    Kevin Kho

    Kevin Kho

    6 months ago
    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
    w

    Wesley Jin

    6 months ago
    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)
    Kevin Kho

    Kevin Kho

    6 months ago
    Glad you figured it out