Can I use Python requests library to trigger a flow run through the GraphQL API?

Archive
, , , ,
1

:point_right:t2:Note that: all GraphQL requests are POST requests rather than GET.

The snippet below creates a flow run with the given parameters.

import requests

create_mutation = """
mutation($input: createFlowRunInput!){
    createFlowRun(input: $input){
        flow_run{
            id
        }
    }
}
"""

inputs = dict(
    versionGroupId="339c86be-5c1c-48f0-b8d3-fe57654afe22", parameters=dict(x=6)
)
response = requests.post(
    url="https://api.prefect.io",
    json=dict(query=create_mutation, variables=dict(input=inputs)),
    headers=dict(authorization=f"Bearer {API_KEY}"),
)
print(response.status_code)
print(response.text)

Using version_group_id ensures that even if you register a new version of the flow, the same code will still work since the version_group_id uniquely identifies a specific flow in a project.

For more on version_group_id, see this page.

2

Trigger flow the event-based execution with AWS Lambda, How to implement this above code in prefect 2, can anyone explain ?

3

hi @md_rafi1 - Prefect 2 is powered by a REST API instead of GraphQL. So if you want to trigger a flow run of a deployment with a http client, you can do something like:

import os

import httpx

api_url = os.getenv('PREFECT_API_URL')
headers = {
  "Authorization": f"Bearer {os.getenv('PREFECT_API_KEY')}"
}
payload = {
  "name": "my-new-flow-run-name", # not required
  #"parameters": {} only required if your flow needs params
}
deployment_id = "<my-deployment-uuid>"

async with httpx.AsyncClient() as client:
  response = await client.post(
    f"{api_url}/deployments/{deployment_id}/create_flow_run",
    headers=headers,
    json=payload
  )
  response.raise_for_status()
4

I have a prefect server deployed in kubernetes, but available in local using port forwarding. For this, how will I obtain PREFECT_API_KEY?