How to trigger a flow run from a deployment via API call using Python requests library or from a terminal using curl?

[anna_geller]

Python requests

Prefect Cloud

Make sure to replace the API_KEY with your API key (e.g., from an environment variable), and the base URL should point to your Prefect Cloud account and workspace.

import requests

API_KEY = "pnu_123456789" 


def create_flow_run(
    deployment_id: str,
    base_url="https://api.prefect.cloud/api/accounts/c5276cbb-62a2-4501-b64a-74d3d900d781/workspaces/aaeffa0e-13fa-460e-a1f9-79b53c05ab36",
):
    return requests.post(
        url=f"{base_url}/deployments/{deployment_id}/create_flow_run",
        json={"name": "your_flow_run", "state": {"type": "SCHEDULED"}},
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {API_KEY}",
        },
    )


if __name__ == "__main__":
    r = create_flow_run("c2fbf9c6-5c96-40a4-bc55-69500f14ed37")
    print(r.status_code)
    print(r.text)

Locally running Prefect 2.0

This post assumes that you have already created a deployment.

Example 1:

import requests


def create_flow_run(deployment_id: str, base_url="http://127.0.0.1:4200/api"):
    return requests.post(
        url=f"{base_url}/deployments/{deployment_id}/create_flow_run",
        json={"name": "your_flow_run", "state": {"type": "SCHEDULED"}},
    )


r = create_flow_run("bd4e3146-8af1-405d-864f-92061a650fd5")
print(r.status_code)
print(r.text)

Alternatively, the same leveraging the data argument:

import json
import requests


def create_flow_run(deployment_id: str, base_url="http://127.0.0.1:4200/api"):
    return requests.post(
        url=f"{base_url}/deployments/{deployment_id}/create_flow_run",
        data=json.dumps(dict(name="xxx", state=dict(type="SCHEDULED"))),
    )


r = create_flow_run("bd4e3146-8af1-405d-864f-92061a650fd5")
print(r.status_code)
print(r.text)

Curl

curl -X POST -H 'Content-Type: application/json' \
http://127.0.0.1:4200/api/deployments/bd4e3146-8af1-405d-864f-92061a650fd5/create_flow_run \
-d '{"name": "curl", "state": {"type": "SCHEDULED"}}'