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

anna_geller · August 16, 2022, 2:05pm

Example flow

from prefect import flow


@flow(log_prints=True)
def main(user: str = "Marvin"):
    print(f"Hello, {user}! 🙌")

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)

Create a deployment, get its ID and trigger a run for it

from prefect.deployments import Deployment
import requests

from hello import main

deploy = Deployment.build_from_flow(flow=main, name="dev", work_queue_name="dev")
id_ = deploy.apply()


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": "Flow run from an API call",
            "state": {"type": "SCHEDULED"},
            "parameters": {"user": "Discourse"},
        },
    )


r = create_flow_run(id_)
print(r.status_code)
print(r.text)

Locally running Prefect 2.0

This example assumes that you have already created a deployment (e.g. created as shown above).

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": "Flow run from an API call",
            "state": {"type": "SCHEDULED"},
            "parameters": {"user": "Discourse"},
        },
    )


r = create_flow_run("02af9c55-cd8c-46ea-b7e1-aa280d76de65")
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"}}'