https://prefect.io logo
e

Eric

09/01/2023, 11:27 PM
Hello, I'm creating my first flow run on local for the first time, and i want to set it up with how (i think) my prod set up will interact with prefect. I have a work pool created and a local agent listening to it, and I deployed my Flow to prefect cloud like so:
Copy code
if __name__ == "__main__":
    
    deployment: Deployment = Deployment.build_from_flow(
        flow=test_flow,
        name="example",
        version="1",
        tags=["demo"],
        work_pool_name='local_agent_work_pool'
    )

    deployment.apply()
I confirmed that the deployment worked in the Prefect Cloud UI. In my flask endpoint, I am trying to submit the flow run to Prefect cloud like so, but it feels wrong:
Copy code
@flask_route.route("/prefect", methods=['POST'])
def start_prefect_flow():
    # prefect_client = get_client(
    #     # api_key='redacted'
    # )
    prefect_client = PrefectClient(
        api='redacted',
        api_key='redacted'
    )

    res = prefect_client.create_flow_run(test_flow)
    print(res)
First, I don't think I should be creating the client this way, but unsure how else I should be passing my api/api key information to it. This is also returning a coroutine object instead of what I thought would be information about my flow run, but maybe that makes sense since I don't see a flow run being kicked off. Some assistance here would be super helpful!
Would also love some recommended best practices for how to manage flows during deployment to prod environments, but that seems like a conversation for after getting through the local dev setup hurdles!
I ended up using httpx.Client and was able to post a request; I was specifically trying to create a flow run from the deployment so the flow run request went to a specific worker pool / worker. What's the best way to do this with the SDK?
n

Nate

09/03/2023, 1:50 AM
hi @Eric -
run_deployment
is the easiest way to do this if you want to use the SDK
Copy code
from prefect.client.schemas.core import FlowRun
from prefect.deployments import run_deployment

# blocks until flow run completes unless you call with `timeout=0`
flow_run: FlowRun = run_deployment("test-flow/test-deployment", params=dict(...))
and if you're just getting started I'd recommend using: • our flow.serve deployment UX if you want a pythonic deployment experience • our worker /
prefect.yaml
deployment UX instead of
Deployment.build_from_flow
- because
build_from_flow
is meant to create block-based deployments that work with agents (not workers) and do not leverage things like
pull
steps etc, and for deployments with dynamically created infra, workers are our main recommendation
e

Eric

09/03/2023, 2:59 PM
Thanks @Nate! I'm still unsure what the best way is to configure Prefect in Python with my required env vars (api url, token), is it to create a client? I've looked over both of those pages you linked multiple times, but neither of them mentioned `run_deployment`; I'm wondering if there is an E2E example using flask or another popular Python WSGI app framework. Running async jobs from an api server seems like a very common case people would want to implement
n

Nate

09/03/2023, 8:39 PM
so anywhere you need to talk to the server via the SDK (cloud or self hosted), you will need to have
PREFECT_API_KEY
and
PREFECT_API_URL
set as env vars, so commonly: • your local machine at deployment time (or github actions env at deployment time) • the machine running your worker (e.g. put it in your
values.yaml
if using the helm chart)
Running async jobs from an api server seems like a very common case people would want to implement
agreed! i think some official examples of this would be helpful, but using
run_deployment
shouldnt be meaningfully different from any simple flask / fastapi endpoint, for example
Copy code
from uuid import UUID

from fastapi import FastAPI, Body
from prefect.deployments import run_deployment

app = FastAPI()

@app.post("/run-prefect-flow/")
async def run_prefect_deployment(slug: str = Body(...), params: dict = Body(...)) -> UUID:
   # timeout = 0 to return immediately instead of blocking
   flow_run =  await run_deployment(slug, parameters=params, timeout=0)
   return flow_run.id

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
which could be invoked like
Copy code
curl -X 'POST' \
  -H 'Content-Type: application/json' \
  --data '{"slug": "foo/some-deployment", "params": {"key": "value"}}' \
  '<http://localhost:8000/run-prefect-flow/>'
note that my
foo/some-deployment
is assumed to already exist and
**params
should match the signature of that deployed flow
🙌 1
e

Eric

09/04/2023, 3:08 AM
Well, i would love to see what the E2E looks like with a deployment; for example, when it is recommended to deploy the flow, how to pass the deployment uuid to the flask app in prod, how to interact with this value in the endpoint etc. Maybe I'm asking for too much to be spoon fed to me 🙂 I will continue working on my prototype app for the time being
n

Nate

09/04/2023, 4:04 PM
yeah E2E looks quite a bit different for everyone, lot of surface area to cover. my example above is just a single way to do this
when it is recommended to deploy the flow, how to pass the deployment uuid to the flask app in prod
these sound a bit like implementation details of your use case that'd be hard to recommend from the outside. happy to discuss further if you have specific questions - that or you could speak with our pro-serv team if that's something you're interested in