Worked example: a nightly report

A report your scheduler runs every night: every issue with activity in an exact window, and, optionally, one saved analytics result that your agent measured earlier. The script writes one JSON file that only its own user can read, and its exit status tells the scheduler whether everything was complete.

The runnable example is scripts/testing/examples/apps/nightly_report.py in the Anectico repository, with the shared client anectico_rest.py. Python 3.11 or later, standard library only.

Before you start

Decision	This example
Where it runs	Your scheduler (cron, a CI schedule, a job runner), once per window.
Project and window	One project, and an explicit --start and --end with a time zone. The script never chooses "recent" for you.
Runtime permissions	errors:read. With --insight, also insights:read. With --result, also analytics:read, persons:read and every source permission the result recorded.
Builder authority	Your agent provisions the key and, separately, runs any measurement whose result the report reads. The script never runs one.
Secret	ANECTICO_API_KEY from the scheduler's secret store.
Expiry and stop	A fixed expiry; anectico apikey revoke <key-id>.
Contract	contract-pin.json; the script writes nothing and exits 3 against another contract.

What it calls

Operation	Qualification	What it means for this report
GET /api/v1/errors/groups	Restricted	The answer states the window it used (effective_start_time, effective_end_time) but not whether that window was cut short by data retention, so the report records the effective window and says retention coverage is not reported. An empty list could be a measured zero or no data; the report never upgrades it to "no errors".
GET /api/v1/analytics/insights/{id}	Supported	The saved definition and its revision. It does not run the insight.
GET /api/v1/analytics/results/{result_id}	Supported	A stored measurement. Reading it starts no execution; current authority and lifecycle still apply.

Your agent fetches exactly these declarations from the deployment before it writes the script:

anectico api spec --partition apps --revision "$(jq -r .contract_revision contract-pin.json)" \
  --operation get_errors_groups --operation get_analytics_insights_by_id \
  --operation get_analytics_results_by_result_id --out nightly-report.contract.json

Every page, with the same selectors

def issues(self) -> dict[str, Any]:
    """Every issue with activity in the window, following the cursor to the last page."""
    selectors = {  # identical on every page; only the cursor changes
        "project_id": self._project,
        "start_time": iso(self._start),
        "end_time": iso(self._end),
        "sort": "last_seen",
        "limit": str(self._page_size),
    }
    by_id: dict[str, dict[str, Any]] = {}
    seen_cursors: set[str] = set()
    cursor, pages, repeats, effective = "", 0, 0, None
    while True:
        query = selectors | ({"cursor": cursor} if cursor else {})
        try:
            answer = self._client.get("/api/v1/errors/groups", query, deadline=60).body
        except (AccessEnded, Forbidden) as ended:
            self.access_ended = True
            return unavailable(ended, pages)
        except AnecticoError as error:
            return unavailable(error, pages)
        pages += 1
        effective = effective or {"start": answer.get("effective_start_time"), "end": answer.get("effective_end_time")}
        for issue in answer["issues"]:
            # Pages are not a frozen snapshot: an issue that moves while the
            # script reads can appear twice. Keep one copy and count repeats.
            if issue["group_id"] in by_id:
                repeats += 1
            by_id[issue["group_id"]] = {
                "group_id": issue["group_id"], "title": issue["title"], "status": issue["status"],
                "service": issue["service_name"], "events": issue["event_count"],
                "affected_people": issue["affected_users"],
                "first_seen": issue.get("first_seen"), "last_seen": issue.get("last_seen"),
            }
        page = answer.get("pagination") or {}
        cursor = page.get("next_cursor") or ""
        if not page.get("has_more") or not cursor:
            break
        if cursor in seen_cursors or pages >= MAX_PAGES:
            return {"state": "incomplete", "reason": "the cursor repeated or the page limit was reached",
                    "pages": pages, "issues": list(by_id.values())}
        seen_cursors.add(cursor)
    return {
        "state": "complete",
        "requested_window": {"start": iso(self._start), "end": iso(self._end)},
        "effective_window": effective,
        "retention_coverage": "not reported by this operation",
        "pages": pages,
        "repeated_on_later_pages": repeats,
        "issue_count": len(by_id),
        "issues": list(by_id.values()),
    }

• The selectors (project, window, sort, page size) are identical on every page; only cursor changes, and it is sent back exactly as received.
• The script stops when has_more is false. A repeated cursor or more than 200 pages ends the component as incomplete, never as a shorter "complete" list.
• Pages are not a frozen snapshot. An issue that changes during the read can appear on two pages; the report keeps one copy and counts the repeats in repeated_on_later_pages.
• pagination.total_count is not used as the number of issues: the report counts what it read.
• Event and person counts are 64-bit decimal strings and are written as strings.

A saved result, read and never re-run

def insight(self, insight_id: str) -> dict[str, Any]:
    path = Client.path("/api/v1/analytics/insights/{id}", id=insight_id)
    try:
        saved = self._client.get(path, {"project_id": self._project}, deadline=30).body["insight"]
    except (AccessEnded, Forbidden) as ended:
        self.access_ended = True
        return unavailable(ended, 0)
    except AnecticoError as error:
        return unavailable(error, 0)
    if saved is None or saved.get("deleted"):
        return {"state": "unavailable", "reason": "the saved insight is deleted"}
    summary = saved["summary"]
    return {"state": "complete", "id": summary["id"], "title": summary["title"], "revision": summary["revision"],
            "definition_hash": saved["definition_hash"]}

def result(self, result_id: str) -> dict[str, Any]:
    """Read a saved result. Polling a pending result runs no new query."""
    path = Client.path("/api/v1/analytics/results/{result_id}", result_id=result_id)
    for poll in range(RESULT_MAX_POLLS):
        try:
            answer = self._client.get(path, {"project_id": self._project}, deadline=30).body
        except (AccessEnded, Forbidden) as ended:
            self.access_ended = True
            return unavailable(ended, 0)
        except AnecticoError as error:
            return unavailable(error, 0)
        status = answer["status"]
        if status == "PRODUCT_RESULT_STATUS_READY":
            kind = next((k for k in PAYLOADS if k in answer), None)
            return {"state": "complete", "status": status, "kind": kind, "payload": answer.get(kind) if kind else None}
        if status in TERMINAL:
            return {"state": "unavailable", "status": status, "failure_reason": answer["failure_reason"],
                    "earliest_available_start": answer.get("earliest_available_start")}
        if poll + 1 < RESULT_MAX_POLLS:
            self._sleep(RESULT_POLL_SECONDS)
    return {"state": "incomplete", "status": status, "reason": "still not ready; read the same result again later"}

A pending result is read again no more than once every five seconds, at most twelve times; reading it creates no new execution. A ready result is copied with its own manifest, which carries its measurement windows and coverage. A failed, expired or invalidated result is reported as unavailable with its status and reason. To report newer numbers, your agent runs a new measurement under a new execution key and the report is given the new result ID; the script never executes one to refresh.

Gaps are marked, never zero

A component that could not be read is written as "state": "unavailable" with the refusal (Busy, AccessEnded, the status, code and Retry-After) and how many pages were read, and it has no issue_count at all. The script exits 2 so the scheduler can alert, and still writes what it did read. The shared client retries a busy or unavailable read with back-off bounded by Retry-After; it never retries a 401 and never switches to another key.

Exit	Meaning
0	Every component is complete.
2	A component is unavailable or incomplete; the report says which.
3	The deployment serves another contract; nothing was read or written.
4	The key's access ended (revoked, expired or narrowed); the report marks it.
5	The output file already exists; nothing was written.
130	Cancelled (SIGTERM or Ctrl-C); nothing was written.

A private output file

def write_private(path: Path, document: dict[str, Any]) -> None:
    """Create the file only if it does not exist, readable by this user alone."""
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w") as out:
        json.dump(document, out, indent=2, sort_keys=True)
        out.write("\n")
        out.flush()
        os.fsync(out.fileno())

The file is created only if it does not exist and is readable by the script's user alone (0600). The report holds no key. It does hold your data (issue titles, service names), so the destination, its retention and its deletion are yours: Anectico's erasure and deletion cannot reach a copy you exported.

Run it

export ANECTICO_API_URL=https://app.anectico.com ANECTICO_PROJECT_ID=<project-uuid>
export ANECTICO_API_KEY="$(your-secret-store read anectico/nightly-report)"
python3 nightly_report.py --start 2026-10-14T00:00:00Z --end 2026-10-15T00:00:00Z \
  --out /var/reports/anectico-2026-10-14.json --result <result-uuid>

Your agent can compare the issues with the CLI, using its own credential:

anectico issues list --since 2026-10-14T00:00:00Z --until 2026-10-15T00:00:00Z -o json

Writes stay out of the nightly run

This report only reads. If you later add an authorized write, follow that operation's own preview, confirmation and state check; after a timeout look the outcome up or repeat the same intent with the same idempotency key, never a new one, and never assume nothing was written.

Stop it and clean up

Revoke the script's key; the next run exits 4 and marks the component. Reports already written stay where you wrote them until you delete them.

Related

• Polling, retries and refusals
• Agent evidence and pagination
• Manage saved insights

Optional conditional revalidation

This one-shot example reads the saved resource once per invocation and keeps no persistent cache. A backend that retains its exact permitted response can send its ETag in If-None-Match on the next saved-definition or ready-result read. A 304 has no JSON body; reuse only the matching body already held for that credential and project. Every request still checks current access, policy, expiry and erasure. Dispose of the cached body and tag after access ends or the result is withdrawn. This saves transfer bytes while stored-payload reads, verification and serialization still run. It starts no execution or accepted-query work. See the full validator syntax and disposal rules.