Dagster - A Dagster+ GraphQL API script for finding runs based on cursor timestamp

find_runs_with_cursor.py

import os

from gql import Client, gql
from gql.transport.requests import RequestsHTTPTransport

url = "https://yourorg.dagster.cloud/prod/graphql"
token = os.getenv(
    "DAGSTER_CLOUD_USER_TOKEN"
)  # a User Token generated from the Cloud Settings page in Dagster Cloud. Note: User Token, not Agent Token

# Define the transport with the endpoint URL and any headers if needed
transport = RequestsHTTPTransport(
    url=url,
    headers={
        "Dagster-Cloud-Api-Token": token,
    },
    use_json=True,
    timeout=120,
)


client = Client(transport=transport)

RUNS_QUERY = gql("""
query RunsQuery($cursor: String) {
  runsOrError(cursor: $cursor, limit: 10) {
    ... on Runs {
     	results {
            id
    	}
    }
    ... on PythonError {
      message
      stack
    }
  }
}
""")

# define the time threshold for what is old enough, this example uses 1 week

cursor = None

while True:
    try:
        result = client.execute(RUNS_QUERY, variable_values={"cursor": cursor})

        if "runsOrError" not in result:
            raise Exception(f"Unexpected result {result}")

        if not result["runsOrError"]["results"]:
            print("No more runs")
            break

        for run in result["runsOrError"]["results"]:
            run_id = run["id"]
            print(f"Found run {run_id}")

        cursor = result["runsOrError"]["results"][-1]["id"]

    except Exception as e:
        print(f"An error occurred: {e}")