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"}}'