Thread
#prefect-server
    Tom Forbes

    Tom Forbes

    1 year ago
    How can I customise the flow executor depending on if the flow is being run locally or being serialized to a docker container? i.e I want to do:
    if running_locally:
        flow.executor = LocalDaskExecutor()
    else:
        flow.executor = DaskExecutor(cluster_class=lambda: KubeCluster(pod_template=....))
    obviously the second DaskExecutor won’t work locally, which is a shame. I can’t find any supported way of checking if we’re serializing this for production rather than debugging it locally
    Kevin Kho

    Kevin Kho

    1 year ago
    Hey @Tom Forbes, do you mean using
    flow.run()
    when you say running locally? Can't you change the executor under a
    if __name__ == "__main__":
    ?
    Tom Forbes

    Tom Forbes

    1 year ago
    Sure,
    flow.run()
    is one way, but it feels a bit unwieldily. We where planning on just having this interface:
    from internal_library import get_executor
    
    with Flow(executor=get_executor()):
       ...
    and have
    get_executor()
    return a different class depending on the context.
    I guess we can just expose another function like
    get_local_executor()
    🤷
    Kevin Kho

    Kevin Kho

    1 year ago
    You're right that this seems to only be doable in script-based storage. I'm not sure a function will work because even if the execution of that is delayed, the executor is already set once that function is run and you won't be able to change it.
    Tom Forbes

    Tom Forbes

    1 year ago
    It would work if we documented that you should do:
    from internal_library import k8s_executor, local_executor
    
    with Flow(executor=k8s_executor()) as flow: 
        ...
    
    if __name__ == "__main__":
        flow.run(executor=local_executor())
    but this kind of sucks as we need to do the same thing for result classes.
    But anyway, thanks for helping!
    Michael Adkins

    Michael Adkins

    1 year ago
    If you store your flow as a script (vs pickling) this is achievable
    prefect.context.get("running_with_backend")
    will be
    True
    when the flow is loaded from storage since 0.14.20
    Before then, you could check for a
    flow_run_id
    ie
    prefect.context.get("flow_run_id")
    Tom Forbes

    Tom Forbes

    1 year ago
    Thanks Michael!
    running_with_backend
    is something I tried, but it’s only available at runtime (which makes sense). I’ll use it to grab the
    Result
    class instead of fiddle with the executor