@Marvin I'm using Prefect 2.19.3, how can I fetch the latest completed flow run for a given flow via the Prefect Server API?

To fetch the latest completed flow run for a given flow using the Prefect Server API, you would typically use the GraphQL API to query the necessary details. Here’s an example of how you can achieve this: 1. Set up GraphQL Query: You need a GraphQL query to fetch the latest completed flow run for a specific flow. The query will look for flow runs associated with a given flow and sort them by

start_time

in descending order, filtering only the completed ones.

graphql
query ($flowId: uuid!) {
  flow_run(where: { flow_id: { _eq: $flowId }, state: { _eq: "COMPLETED" } }, order_by: { start_time: desc }, limit: 1) {
    id
    name
    start_time
    end_time
    state
  }
}

2. Execute the Query: You can execute this query using an HTTP client in Python to interact with the Prefect Server API. Here’s an example using the

requests

library:

python
import requests

url = "<http://your-prefect-server/api>"
headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer YOUR_PREFECT_API_KEY"
}

query = """
query ($flowId: uuid!) {
  flow_run(where: { flow_id: { _eq: $flowId }, state: { _eq: "COMPLETED" } }, order_by: { start_time: desc }, limit: 1) {
    id
    name
    start_time
    end_time
    state
  }
}
"""

variables = {
    "flowId": "your-flow-id"  # Replace with your actual flow ID
}

response = <http://requests.post|requests.post>(url, json={'query': query, 'variables': variables}, headers=headers)

if response.status_code == 200:
    data = response.json()
    latest_completed_run = data['data']['flow_run']
    print("Latest Completed Flow Run:", latest_completed_run)
else:
    print(f"Query failed to run with a {response.status_code}.")

Note: Replace

<http://your-prefect-server>

with the actual URL of your Prefect server, and replace

YOUR_PREFECT_API_KEY

with your actual Prefect API key. That should fetch the latest completed flow run for the specified flow.

I don't need to point this query at any specific endpoint?

It looks like you might be pointing to an endpoint that doesn't accept POST requests. For Prefect Server, GraphQL queries should be sent to the

/graphql

endpoint. Here's how you can adjust your request to point to the correct endpoint:

python
import requests

url = "<http://your-prefect-server/graphql>"  # Use /graphql endpoint
headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer YOUR_PREFECT_API_KEY"  # Include your Prefect API key if required
}

query = """
query ($flowId: uuid!) {
  flow_run(where: { flow_id: { _eq: $flowId }, state: { _eq: "COMPLETED" } }, order_by: { start_time: desc }, limit: 1) {
    id
    name
    start_time
    end_time
    state
  }
}
"""

variables = {
    "flowId": "your-flow-id"  # Replace with your actual flow ID
}

response = <http://requests.post|requests.post>(url, json={'query': query, 'variables': variables}, headers=headers)

if response.status_code == 200:
    data = response.json()
    latest_completed_run = data['data']['flow_run']
    print("Latest Completed Flow Run:", latest_completed_run)
else:
    print(f"Query failed to run with a {response.status_code}.")

Make sure your Prefect server's DNS or IP and port are correctly specified in the

url

variable. This endpoint should process the GraphQL query appropriately.

@Marvin can I do this using the Prefect Server REST API instead of using GraphQL? It doesn't seem to be working with the latter

Alright, it seems we’re hitting some issues with the documentation search. Let's try to outline how you could use the Prefect REST API to fetch the latest completed flow run. Below is a method using the Prefect REST API. Using the Prefect REST API: 1. Retrieve the Flow ID: Fetch the flow ID for your target flow if you don't already have it.

python
import requests

# Set up the necessary URLs and headers
base_url = "<http://your-prefect-server/api>"
headers = {
    "Authorization": "Bearer YOUR_PREFECT_API_KEY"  # Include your Prefect API key if required
}

# Retrieve the flow ID
flow_name = "your-flow-name"
flows_url = f"{base_url}/flows"
flows_response = requests.get(flows_url, headers=headers)
flows = flows_response.json()

flow_id = None
for flow in flows:
    if flow['name'] == flow_name:
        flow_id = flow['id']
        break

if flow_id is None:
    raise ValueError("Flow ID not found for the specified flow name.")

print(f"Flow ID: {flow_id}")

2. Fetch the Latest Completed Flow Run: Use the obtained flow ID to fetch the latest completed flow run.

python
# Retrieve completed flow runs for the given flow ID
flow_runs_url = f"{base_url}/flow_runs"
params = {
    "flow_id": flow_id,
    "state": "COMPLETED",
    "order_by": "-start_time",
    "limit": 1
}
flow_runs_response = requests.get(flow_runs_url, params=params, headers=headers)
flow_runs = flow_runs_response.json()

if not flow_runs:
    print("No completed flow runs found for the given flow.")
else:
    latest_completed_flow_run = flow_runs[0]
    print("Latest Completed Flow Run:", latest_completed_flow_run)

Note: Replace placeholders like

<http://your-prefect-server>

and

YOUR_PREFECT_API_KEY

with your actual server address and API key. This should allow you to interact with the Prefect REST API to retrieve the latest completed flow run for a specified flow. Ensure that your Prefect server's REST API is correctly configured to accept these requests.

@Marvin this is also unfortunately returning

Method Not Allowed

It's evident that accessing the Prefect REST API can be tricky with the responses pointing out

Method Not Allowed

. The likely cause is using the wrong endpoint or method. Let's ensure we're using the correct REST API path and methods. Here's a revised approach: 1. Correct REST API Endpoints: Prefect's REST endpoints can vary, so it's crucial to double-check that you're using the right ones. Typically, you need to interact with

/flow_runs/

for operations related to flow runs. 2. Proper HTTP Methods: Ensure you use the correct HTTP methods (

GET

,

POST

, etc.) according to what the endpoint specifications allow. Here's an example to fetch the latest completed flow run using a more structured approach: Fetch Flow ID:

python
import requests

# Set up the necessary URLs and headers
base_url = "<http://your-prefect-server/api>"
headers = {
    "Authorization": "Bearer YOUR_PREFECT_API_KEY"  # Include your Prefect API key if required
}

# Retrieve the flow ID
flow_name = "your-flow-name"  # Replace with your actual flow name
flows_url = f"{base_url}/flows"

params = {
    "name": flow_name
}

flows_response = requests.get(flows_url, params=params, headers=headers)

if flows_response.status_code == 200:
    flows = flows_response.json()
    if flows:
        flow_id = flows[0]['id']
        print(f"Flow ID: {flow_id}")
    else:
        print("No flow found with the specified name.")
        flow_id = None
else:
    print(f"Failed to fetch flows: {flows_response.status_code}")
    flow_id = None

Fetch Latest Completed Flow Run:

python
if flow_id:
    # Retrieve completed flow runs for the given flow ID
    flow_runs_url = f"{base_url}/flow_runs"
    params = {
        "flow_id": flow_id,
        "state": "COMPLETED",
        "order_by": "-start_time",
        "limit": 1
    }
    flow_runs_response = requests.get(flow_runs_url, params=params, headers=headers)
    
    if flow_runs_response.status_code == 200:
        flow_runs = flow_runs_response.json()
        if flow_runs:
            latest_completed_flow_run = flow_runs[0]
            print("Latest Completed Flow Run:", latest_completed_flow_run)
        else:
            print("No completed flow runs found for the given flow.")
    else:
        print(f"Failed to fetch flow runs: {flow_runs_response.status_code}")
else:
    print("Cannot fetch flow runs without a valid flow ID.")

Make sure: 1. The

base_url

is accurate and points correctly to your Prefect Server instance. 2. The

flow_name

accurately reflects an existing flow in your Prefect instance. 3. The API key is correctly set up if required by your server configuration. Validate the available endpoints and methods in your Prefect Server by referring to your server's API documentation or swagger UI if available. This approach ensures you're using correct endpoints and parameters. If you still encounter

Method Not Allowed

, your server might have specific configurations or additional security constraints.

sorry about this @Del - i think something is wrong with marvin's retrieval right now, will take a look https://docs.prefect.io/latest/guides/using-the-client/#get-the-last-n-completed-flow-runs-from-my-workspace

Thanks, this helps