Export Incident Groups to CSV

Steps

Get incident groups for a page

Create a function that takes a client, a source ID, a cursor, and a page limit. The function sends the query for that cursor, returning up to page_limit incident groups. By default, end_cursor is None, which starts from the first record.

Create a CSV writer

Create a CSV writer, and write the header to the file.

Loop over all pages

Start a loop that runs as long as the API reports another page. Each response has an end cursor, which you pass to get_incident_groups as after so the next page starts from that point. For more on how pagination works, see Pagination and filtering.

Write the incidents to file

Iterate over all incident groups on the page and write their incidents to the CSV file.

Full code sample

Python

import csv
import os
from pathlib import Path

from validio_sdk.client import Client


async def get_incident_groups(
    client: Client,
    source_id: str,
    end_cursor: str | None = None,
    page_limit: int = 10,
):
    query = """
    query IncidentGroups(
        $filter: IncidentGroupsFilter,
        $pagination: PaginationInput,
        $sort: IncidentGroupsSort
    ) {
      incidentGroups(
          filter: $filter,
          pagination: $pagination,
          sort: $sort
        ) {
          elements {
            firstSeenAt
            lastSeenAt
            incidents {
              elements {
                startTime
                endTime
                severity
                value
                deviation
              }
            }
          }
          pageInfo {
            endCursor
            hasNextPage
          }
        }
      }
    """

    variable_values = {
        "filter": {
            "source": {"values": [source_id]},
            "status": {"values": ["TRIAGE"]},
        },
        "sort": {
            "firstSeenAt": {
                "index": 1,
                "order": "ASCENDING",
            },
        },
        "pagination": {
            "after": end_cursor,
            "limit": page_limit,
        },
    }

    return await client.execute(query, variable_values=variable_values)


async def example(source_id: str = os.environ["EXAMPLE_SOURCE_ID"]):
    client = Client()
    end_cursor = None
    has_next_page = True

    with Path("incidents.csv").open("w", encoding="utf-8") as csv_file:
        writer = csv.DictWriter(
            csv_file,
            fieldnames=[
                "startTime",
                "endTime",
                "severity",
                "value",
                "deviation",
            ],
        )

        writer.writeheader()

        while has_next_page:
            result = (await get_incident_groups(client, source_id, end_cursor))[
                "incidentGroups"
            ]
            has_next_page = result["pageInfo"]["hasNextPage"]
            end_cursor = result["pageInfo"]["endCursor"]

            for incident_group in result["elements"]:
                writer.writerows(incident_group["incidents"]["elements"])

Overview

Getting Started

Validio API

Validio CLI

Validio SDK

Validio Code (IaC)

Overview

Steps

Full code sample

​Overview

​Steps

​Full code sample

Overview

Steps

Full code sample