Dagster - A Dagster+ GraphQL API script for finding a run's step data using its ID

find_run_step_data.py

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

# Define the endpoint URL and token
org_name = "your_org_name"
base_url = f"https://{org_name}.dagster.cloud/"
deployment_name = "your_deployment_name" # branch deployments use an ID
url = base_url + deployment_name + "/graphql"
token = os.getenv("DAGSTER_CLOUD_USER_TOKEN")  # a User Token generated from the Organization settings page in Dagster+. 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,
)

# Instantiate the client
client = Client(transport=transport)

# Define the GraphQL query
RUN_QUERY = gql("""
query RunQuery($runId: ID!) {
  runOrError(runId: $runId) {
    __typename
    ... on Run {
      id
      status
      startTime
      endTime
      stepStats {
        stepKey
        startTime
        endTime
        status
      }
    }
    ... on PythonError {
      message
      stack
    }
  }
}
""")

# Define the run ID you want to query
run_id = "your_run_id_here"

# Execute the query
try:
    result = client.execute(RUN_QUERY, variable_values={"runId": run_id})
    if "runOrError" not in result:
        raise Exception(f"Unexpected result {result}")

    run_or_error = result["runOrError"]
    if run_or_error["__typename"] == "Run":
        print(f"Run ID: {run_or_error['id']}")
        print(f"Status: {run_or_error['status']}")
        print(f"Start Time: {run_or_error['startTime']}")
        print(f"End Time: {run_or_error['endTime']}")
        print("Step Stats:")
        for step in run_or_error["stepStats"]:
            print(f"  Step Key: {step['stepKey']}")
            print(f"  Start Time: {step['startTime']}")
            print(f"  End Time: {step['endTime']}")
            print(f"  Status: {step['status']}")
    elif run_or_error["__typename"] == "PythonError":
        print(f"Error Message: {run_or_error['message']}")
        print(f"Stack Trace: {run_or_error['stack']}")
    else:
        print(f"Unexpected typename: {run_or_error['__typename']}")

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