← All code samples

SRE → pull alerts

Refresh the SRE inbox from Grafana (or another configured source), wait until the pull finishes, then list active alerts.

Step What the SDK does
1. Start pull pull_alerts (joins an in-progress pull on HTTP 409)
2. Wait wait_for_alert_pull / pull_alerts_and_wait
3. List list_alerts

Use this before investigate alert when the inbox may be stale.

Minimal

from stackgen import StackgenClient, StackgenConfig

client = StackgenClient(
    StackgenConfig(
        base_url="https://app.stackgen.com",
        api_token="stackgen_…",
        workspace_id="<workspace-uuid>",
    )
)

pull = client.sre.pull_alerts_and_wait(source="grafana")
print(pull.get("alerts_created"), pull.get("alerts_updated"), pull.get("alerts_resolved"))

alerts = client.sre.list_alerts(status="active", page_size=10)
for row in alerts.get("alerts") or []:
    print(row["id"], row.get("title"), row.get("attention"))

How to run

export STACKGEN_URL=https://app.stackgen.com
export STACKGEN_TOKEN=stackgen_…
export STACKGEN_WORKSPACE_ID=<workspace-uuid>

python examples/sre/pull-alerts.py
# or start only: python examples/sre/pull-alerts.py --no-wait

Complete script

#!/usr/bin/env python3
"""SRE: pull fresh alerts, then list the active inbox.

Documented at:
  https://appcd-dev.github.io/stackgen-sdk/code-samples/sre-pull-alerts/

Credentials (flags override env):
  STACKGEN_URL, STACKGEN_TOKEN, STACKGEN_WORKSPACE_ID (or STACKGEN_ORG_ID / STACKGEN_PROJECT)

Usage:

  python examples/sre/pull-alerts.py --no-wait

  python examples/sre/pull-alerts.py --source grafana
"""

from __future__ import annotations

import argparse
import json
import os
import sys

from stackgen import StackgenClient, StackgenConfig
from stackgen.errors import StackgenError


def _org_id(cli_value: str | None) -> str:
    return (
        (cli_value or "").strip()
        or os.environ.get("STACKGEN_WORKSPACE_ID", "").strip()
        or os.environ.get("STACKGEN_ORG_ID", "").strip()
        or os.environ.get("STACKGEN_PROJECT", "").strip()
        or os.environ.get("STACKGEN_PROJECT_ID", "").strip()
    )


def _config(args: argparse.Namespace) -> StackgenConfig:
    base = (args.base_url or os.environ.get("STACKGEN_URL", "")).strip()
    token = (args.api_token or os.environ.get("STACKGEN_TOKEN", "")).strip()
    org = _org_id(args.workspace_id)
    if not base or not token:
        raise SystemExit(
            "set --base-url/--api-token or STACKGEN_URL / STACKGEN_TOKEN"
        )
    return StackgenConfig(
        base_url=base,
        api_token=token,
        workspace_id=org,
        timeout_seconds=args.timeout_seconds,
        poll_interval_seconds=args.poll_interval_seconds,
    )


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", default="")
    parser.add_argument("--api-token", default="")
    parser.add_argument("--workspace-id",
        "--org-id", default="")
    parser.add_argument("--source", default="grafana")
    parser.add_argument("--integration", default="")
    parser.add_argument(
        "--no-wait",
        action="store_true",
        help="Start the pull and exit without waiting",
    )
    parser.add_argument("--timeout-seconds", type=int, default=1800)
    parser.add_argument("--poll-interval-seconds", type=int, default=5)
    args = parser.parse_args()

    client = StackgenClient(_config(args))
    integration = args.integration.strip() or None

    if args.no_wait:
        started = client.sre.pull_alerts(source=args.source, integration=integration)
        print("pull:", json.dumps(started, indent=2, default=str))
        return 0

    finished = client.sre.pull_alerts_and_wait(
        source=args.source,
        integration=integration,
    )
    print("pull:", json.dumps(finished, indent=2, default=str))
    alerts = client.sre.list_alerts(status="active", page_size=10)
    print("alerts:", json.dumps(alerts, indent=2, default=str))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except StackgenError as exc:
        print(f"error: {exc}", file=sys.stderr)
        if "401" in str(exc) or "missing identity" in str(exc).lower():
            print(
                "hint: SRE /app/sre routes need the edge to inject workspace "
                "identity (X-Appcd-Org). If Aiden works but SRE returns 401, "
                "the mothership app proxy is not forwarding PAT identity.",
                file=sys.stderr,
            )
        raise SystemExit(1)

Notes

  • A pull is async. The start call returns sync_run_id with status running.
  • If another pull is already running for the same source and integration, the server returns 409 with that run. The SDK returns it so you can keep waiting.
  • Pass integration when the workspace has more than one Grafana binding.

Related: Investigate alert · Triage alerts