CI/CD with AnomalyGuard CLI

Use the anomalyguard CLI (or the same REST calls) in a pipeline to version a data view YAML in Git, apply it to AnomalyGuard, deploy, enable, and enqueue load/process jobs.

What the API accepts

The Web UI edits a YAML definition (schemaVersion: 3). The REST API create/update endpoints expect a Fact JSON body (POST/PUT /api/v1/dataviews). The UI converts YAML → Fact when you save; in CI you do the same conversion, then call the API.

Artifact	Role
*.yaml in Git	Source of truth for humans / PRs
*.fact.json (generated)	Payload for dataviews create\|update
Admin API key	Pipeline auth (ANOMALYGUARD_API_KEY)

See Data Views — YAML definition and CI/CD and API Keys.

End-to-end flow

Git (YAML)
  → convert to Fact JSON
    → create or update via API/CLI
      → deploy (first time / after undeploy)
        → enable
          → jobs-custom-load (history) and/or jobs-load
            → jobs-process
              → monitor status

Prerequisites

• Connector named in YAML already exists in AnomalyGuard (same source.connector string).
• Admin API key (create/update/deploy require admin).
• CLI version matches the server (How to use the CLI).

1. Repo layout

dataviews/
  sales/
    sales_by_channel.yaml          # edited in PRs
  scripts/
    yaml_to_fact.py                # YAML → Fact JSON
    deploy_dataview.sh             # convert + create/update + deploy + jobs
.github/workflows/deploy-dataview.yml

2. Sample YAML in Git

dataviews/sales/sales_by_channel.yaml (abbreviated but valid shape):

schemaVersion: 3
metadata:
  name: sales_by_channel
  domain: sales
  description: Daily sales by country and channel
  tags: retail, daily
  dataSource: ERP
  enabled: false          # pipeline enables after successful deploy
source:
  connector: warehouse_pgsql
  dateColumn: order_date
  valueColumn: amount
  countColumn: count
  loadMode: incremental
  cron: "0 6 * * *"
  loadQuery: |
    SELECT CAST(to_char(order_date, 'YYYYMMDD') AS int) AS order_date,
           country, channel, amount, 1 AS count
    FROM sales.daily_facts
    WHERE order_date = CURRENT_DATE - 1
  customLoadQuery: |
    SELECT CAST(to_char(order_date, 'YYYYMMDD') AS int) AS order_date,
           country, channel, amount, 1 AS count
    FROM sales.daily_facts
    WHERE order_date >= CURRENT_DATE - 365
categories:
  - name: country
    label: Country
  - name: channel
    label: Sales channel
    parent: country
analysis:
  fillGapMethod: zero
  ignoreLastNDays: 1
  scanInLastNDays: 1095
  detectors:
    - type: dod
      params:
        percentualChange: 40.0
        mode: both
        useSpecialDates: true
    - type: gap
      params:
        longerThan: 3
        shorterThan: 0
  specialDates:
    - "2026-01-01"
  recurringRules:
    - "month_end"

Export the same document from the UI with YAML definition (</>) if you start from an existing data view.

3. Convert YAML → Fact JSON

scripts/yaml_to_fact.py — maps schemaVersion: 3 YAML to the Fact JSON the API expects (same idea as the UI mapper):

#!/usr/bin/env python3
"""Convert AnomalyGuard data-view YAML (schemaVersion 3) to Fact JSON for the API."""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

import yaml  # pip install pyyaml


def detectors_yaml_blob(analysis: dict) -> str:
    blob: dict = {"detectors": analysis.get("detectors") or []}
    if analysis.get("specialDates"):
        blob["specialDates"] = analysis["specialDates"]
    if analysis.get("recurringRules"):
        blob["recurringRules"] = analysis["recurringRules"]
    if analysis.get("specialPeriods"):
        blob["specialPeriods"] = analysis["specialPeriods"]
    return yaml.safe_dump(blob, sort_keys=False)


def to_fact(doc: dict, existing_id: int | None = None) -> dict:
    meta = doc.get("metadata") or {}
    source = doc.get("source") or {}
    analysis = doc.get("analysis") or {}

    name = (meta.get("name") or "").strip().lower()
    domain = (meta.get("domain") or "").strip().lower()
    if not name or not domain:
        raise SystemExit("metadata.name and metadata.domain are required")

    categories = []
    for c in doc.get("categories") or []:
        categories.append(
            {
                "name": (c.get("name") or "").strip().lower(),
                "label": (c.get("label") or c.get("name") or "").strip(),
                "parentName": (
                    None
                    if not (c.get("parent") or "").strip()
                    else c["parent"].strip().lower()
                ),
                "description": "",
                "isGlobal": False,
            }
        )

    fact = {
        "name": name,
        "domain": domain,
        "description": (meta.get("description") or "").strip() or None,
        "tags": (meta.get("tags") or "").strip(),
        "factSource": (meta.get("dataSource") or "").strip(),
        "enabled": False,  # enable explicitly after deploy
        "connectorName": (source.get("connector") or "").strip().lower(),
        "dateColumn": (source.get("dateColumn") or "").strip().lower(),
        "sumColumn": (source.get("valueColumn") or "").strip().lower(),
        "countColumn": (source.get("countColumn") or "").strip().lower(),
        "loadMode": (source.get("loadMode") or "incremental").strip().lower(),
        "cronTab": (source.get("cron") or "").strip(),
        "loadQuery": (source.get("loadQuery") or "").strip(),
        "customLoadQuery": (source.get("customLoadQuery") or "").strip(),
        "fillGapMethod": (analysis.get("fillGapMethod") or "zero").strip(),
        "ignoreLastNDays": int(analysis.get("ignoreLastNDays") or 0),
        "scanInLastNDays": int(analysis.get("scanInLastNDays") or 0),
        "analyzedVariables": "count,value,avg",
        "detectorsYaml": detectors_yaml_blob(analysis),
        "categories": categories,
    }

    if existing_id is not None:
        fact["id"] = existing_id

    return fact


def main() -> None:
    p = argparse.ArgumentParser()
    p.add_argument("yaml_path")
    p.add_argument("-o", "--output", required=True)
    p.add_argument("--id", type=int, default=None, help="Set Fact.id for updates")
    args = p.parse_args()

    doc = yaml.safe_load(Path(args.yaml_path).read_text(encoding="utf-8"))
    if not isinstance(doc, dict):
        raise SystemExit("YAML root must be a mapping")

    fact = to_fact(doc, existing_id=args.id)
    Path(args.output).write_text(json.dumps(fact, indent=2), encoding="utf-8")
    print(f"Wrote {args.output} ({fact['domain']}/{fact['name']})", file=sys.stderr)


if __name__ == "__main__":
    main()

pip install pyyaml
python scripts/yaml_to_fact.py dataviews/sales/sales_by_channel.yaml \
  -o /tmp/sales_by_channel.fact.json

4. Deploy script (create or update + lifecycle)

scripts/deploy_dataview.sh:

#!/usr/bin/env bash
# Apply a Git YAML data view via AnomalyGuard CLI, then deploy and run jobs.
set -euo pipefail

YAML_PATH="${1:?Usage: $0 <path-to-dataview.yaml> [--skip-jobs]}"
SKIP_JOBS="${2:-}"

: "${ANOMALYGUARD_BASE_URL:?Set ANOMALYGUARD_BASE_URL}"
: "${ANOMALYGUARD_API_KEY:?Set ANOMALYGUARD_API_KEY (admin)}"

command -v anomalyguard >/dev/null
command -v jq >/dev/null
command -v python3 >/dev/null

FACT_JSON="$(mktemp)"
trap 'rm -f "$FACT_JSON"' EXIT

python3 scripts/yaml_to_fact.py "$YAML_PATH" -o "$FACT_JSON"

DOMAIN="$(jq -r '.domain' "$FACT_JSON")"
NAME="$(jq -r '.name' "$FACT_JSON")"

echo "== ping =="
anomalyguard ping --json | jq -e '.statusCode == 200'

echo "== lookup existing $DOMAIN / $NAME =="
LIST_JSON="$(anomalyguard dataviews list --domain "$DOMAIN" --name "$NAME" --json)"
EXISTING_ID="$(echo "$LIST_JSON" | jq -r '
  (.body | fromjson)
  | if type == "array" then
      map(select(.domain == "'"$DOMAIN"'" and .name == "'"$NAME"'")) | .[0].id // empty
    else
      .id // empty
    end
')"

if [[ -n "$EXISTING_ID" && "$EXISTING_ID" != "null" ]]; then
  echo "Updating data view id=$EXISTING_ID"
  python3 scripts/yaml_to_fact.py "$YAML_PATH" -o "$FACT_JSON" --id "$EXISTING_ID"
  # Preserve server-side flags that must match route/body on update
  FULL="$(anomalyguard dataviews get "$EXISTING_ID" --json | jq '.body | fromjson')"
  jq --argjson full "$FULL" '
    .id = $full.id
    | .deployed = $full.deployed
    | .created = $full.created
    | .lastLoad = $full.lastLoad
    | .startDate = $full.startDate
    | .endDate = $full.endDate
  ' "$FACT_JSON" > "${FACT_JSON}.merged"
  mv "${FACT_JSON}.merged" "$FACT_JSON"

  anomalyguard dataviews update "$EXISTING_ID" --file "$FACT_JSON" --json \
    | jq -e '.statusCode == 200'
  DV_ID="$EXISTING_ID"
else
  echo "Creating new data view"
  CREATE_OUT="$(anomalyguard dataviews create --file "$FACT_JSON" --json)"
  echo "$CREATE_OUT" | jq -e '.statusCode == 200'
  DV_ID="$(echo "$CREATE_OUT" | jq -r '.body | fromjson | .id')"
  echo "Created id=$DV_ID"
fi

echo "== deploy =="
# Safe if already deployed: API may no-op or return a clear error — adjust if your server rejects re-deploy.
DEPLOYED="$(anomalyguard dataviews get "$DV_ID" --json | jq -r '.body | fromjson | .deployed')"
if [[ "$DEPLOYED" != "true" ]]; then
  anomalyguard dataviews deploy "$DV_ID" --json | jq -e '.statusCode == 200'
else
  echo "Already deployed — definition updated in place (categories/name/domain still locked)."
fi

echo "== enable =="
anomalyguard dataviews enable "$DV_ID" --json | jq -e '.statusCode == 200'

if [[ "$SKIP_JOBS" == "--skip-jobs" ]]; then
  echo "Skipping jobs."
  exit 0
fi

echo "== enqueue jobs =="
# First-time / history: custom load when customLoadQuery is set; else daily load.
HAS_CUSTOM="$(jq -r '.customLoadQuery | length > 0' "$FACT_JSON")"
if [[ "$HAS_CUSTOM" == "true" ]]; then
  anomalyguard dataviews jobs-custom-load "$DV_ID" --json | jq -e '.statusCode == 200'
else
  anomalyguard dataviews jobs-load "$DV_ID" --json | jq -e '.statusCode == 200'
fi

anomalyguard dataviews jobs-process "$DV_ID" --json | jq -e '.statusCode == 200'

echo "== status =="
anomalyguard dataviews status "$DV_ID" --json | jq '.body | fromjson'

echo "Done. dataViewId=$DV_ID ($DOMAIN/$NAME)"

Make it executable: chmod +x scripts/deploy_dataview.sh.

Category / name / domain changes: those fields are locked after deploy. To change them, run anomalyguard dataviews undeploy {id} (drops tables), update YAML, then deploy again — do not do that lightly in production CI.

5. Same flow with raw HTTP (API)

export ANOMALYGUARD_BASE_URL="https://anomalyguard.contoso.com"
export ANOMALYGUARD_API_KEY="<admin-api-key>"

python3 scripts/yaml_to_fact.py dataviews/sales/sales_by_channel.yaml -o /tmp/dv.fact.json

# Create
curl -fsS -X POST "$ANOMALYGUARD_BASE_URL/api/v1/dataviews" \
  -H "X-API-Key: $ANOMALYGUARD_API_KEY" \
  -H "Content-Type: application/json" \
  --data-binary @/tmp/dv.fact.json

# Or update (id must match body.id)
curl -fsS -X PUT "$ANOMALYGUARD_BASE_URL/api/v1/dataviews/5" \
  -H "X-API-Key: $ANOMALYGUARD_API_KEY" \
  -H "Content-Type: application/json" \
  --data-binary @/tmp/dv.fact.json

curl -fsS -X POST "$ANOMALYGUARD_BASE_URL/api/v1/dataviews/5/deploy" \
  -H "X-API-Key: $ANOMALYGUARD_API_KEY"

curl -fsS -X POST "$ANOMALYGUARD_BASE_URL/api/v1/dataviews/5/enable" \
  -H "X-API-Key: $ANOMALYGUARD_API_KEY"

curl -fsS -X POST "$ANOMALYGUARD_BASE_URL/api/v1/dataviews/sales/sales_by_channel/jobs/custom-load" \
  -H "X-API-Key: $ANOMALYGUARD_API_KEY"

curl -fsS -X POST "$ANOMALYGUARD_BASE_URL/api/v1/dataviews/sales/sales_by_channel/jobs/process" \
  -H "X-API-Key: $ANOMALYGUARD_API_KEY"

6. GitHub Actions sample

# .github/workflows/deploy-dataview.yml
name: Deploy data view

on:
  push:
    branches: [main]
    paths:
      - "dataviews/**/*.yaml"
      - "scripts/**"
  workflow_dispatch:
    inputs:
      yaml_path:
        description: Path to data view YAML
        required: true
        default: dataviews/sales/sales_by_channel.yaml

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install CLI
        run: |
          curl -fsSL -o anomalyguard-linux-x64.tar.gz \
            "${{ secrets.ANOMALYGUARD_BASE_URL }}/api/cli/download/ubuntu"
          tar -xzf anomalyguard-linux-x64.tar.gz
          sudo mv anomalyguard /usr/local/bin/
          anomalyguard --version

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install deps
        run: pip install pyyaml

      - name: Deploy from Git YAML
        env:
          ANOMALYGUARD_BASE_URL: ${{ secrets.ANOMALYGUARD_BASE_URL }}
          ANOMALYGUARD_API_KEY: ${{ secrets.ANOMALYGUARD_API_KEY }}
        run: |
          YAML="${{ github.event.inputs.yaml_path || 'dataviews/sales/sales_by_channel.yaml' }}"
          chmod +x scripts/deploy_dataview.sh
          ./scripts/deploy_dataview.sh "$YAML"

Store ANOMALYGUARD_BASE_URL and ANOMALYGUARD_API_KEY as repository secrets. Prefer an admin key dedicated to CI; rotate when the pipeline is retired.

7. Verify after deploy

anomalyguard dataviews status "$DV_ID" --json | jq '.body | fromjson'
anomalyguard dataviews list --domain sales --name sales_by_channel --json

# Optional gate on anomaly volume for a known filter
COUNT=$(anomalyguard anomalies by-filter "$FILTER_ID" --json \
  | jq '.body | fromjson | length')
echo "Anomalies for filter $FILTER_ID: $COUNT"

Watch job progress in Monitoring.

Tips

• Prefer API keys scoped carefully; deploy lifecycle needs Admin.
• Pin CLI version to the deployed AnomalyGuard release.
• Keep enabled: false in Git YAML and let the pipeline call enable after a successful deploy.
• Changing categories, name, or domain requires undeploy first — treat as a breaking change in the PR review.
• Use anomalyguard api only when no dedicated subcommand exists.
• Keep a backup of the external PostgreSQL database that holds AnomalyGuard state.

Related

• Data Views
• How to use the CLI
• How to use the API
• API Keys
• Monitoring
• AI Agents & AnomalyGuard