Does anyone know if it is possible to figure out in which context a flow is executing, local or remote - and accordingly modifying a task runner? I have a Ray cluster in k8s that our remote deployments will use, but have a requirement to allow local exec too. Problem is, users won't have access to our Ray cluster, which wouldn't be a problem, but the flow def has the address hardcoded. Looking to do something like (local exec has a temp Ray cluster, k8s exec has remote cluster):

@flow(
    task_runner=RayTaskRunner(
        address="<ray://ray-cluster-kuberay-head-svc:10001>"
    ),
)
def main(...):
    ...

if __name__ == "__main__": # If this flow is called directly, don't use the address
    main(..., task_runner=RayTaskRunner)

You can check for a flow run id in the environment to see if it’s being run via deployment

ha, that'll do it. Any idea about modding the flow task runner accordingly?

Would something like this work for your use case when running the flow directly?

if __name__ == "__main__":
  flow_run = main.with_options(task_runner=RayTaskRunner())
  flow_run()

oo I hadn't seen

with_options

, will give that a try

Oh yeah that’d do it too.

working 🙂 Thank you both 👍

We execute your script to load the flow, so you can do

def get_task_runner():
   if "PREFECT__FLOW_RUN_ID" in os.environ:
       return RayTaskRunner(
        address="<ray://ray-cluster-kuberay-head-svc:10001>"
       )
    else:
      ...
@flow(
    task_runner=get_task_runner()
)

Ha, good point 😄