Rapportage use cases

Deze voorbeelden laten zien hoe je de operationele rapporten bouwt die compliance teams vaak nodig hebben: control-effectiviteit per periode, overzichten van overdue tasks, en gecombineerde statusrapporten.

Info

De API ondersteunt geen server-side datumfilters. Alle periodefiltering (bijv. "alleen februari", "Q1") doe je client-side door te filteren op het closedDate-veld nadat je de data hebt opgehaald.

Tip

De scripts op deze pagina gebruiken get_token() van de pagina Authenticatie. Kopieer die functie naar je script, of importeer hem vanuit een gedeelde module.

Control-effectiviteitsrapport (per periode)

Assessments zijn de bron voor effectiviteit. Elke assessment heeft een closedDate (wanneer de taak is afgerond) en een effectiveness-waarde (EFFECTIVE, INEFFECTIVE of UNDETERMINED).

Om te rapporteren over een specifieke maand of kwartaal, haal je alle assessments op en filter je op closedDate in je script.

import requests
import csv
from datetime import date

GRAPHQL_URL = "https://portal.tidalcontrol.com/graphql"

QUERY = """
query ExportAssessments($first: Int, $after: String, $filter: ActivityFilter) {
  activities_paged(first: $first, after: $after, filter: $filter) {
    edges {
      node {
        id
        sequenceId
        name
        closedDate
        createdDate
        controls { id sequenceId name category }
        assets { id name }
        assignments { assignmentType user { name email } }
        ... on Assessment {
          effectiveness
        }
      }
    }
    pageInfo { hasNextPage endCursor }
  }
}
"""


def graphql(query, variables, access_token):
    r = requests.post(
        GRAPHQL_URL,
        json={"query": query, "variables": variables},
        headers={"Authorization": f"Bearer {access_token}"},
    )
    r.raise_for_status()
    result = r.json()
    if "errors" in result:
        raise RuntimeError(result["errors"])
    return result["data"]


def fetch_all_assessments(access_token):
    results = []
    cursor = None
    while True:
        data = graphql(QUERY, {"first": 50, "after": cursor, "filter": {"activityTypes": ["ASSESS"]}}, access_token)
        page = data["activities_paged"]
        results.extend(edge["node"] for edge in page["edges"])
        if not page["pageInfo"]["hasNextPage"]:
            break
        cursor = page["pageInfo"]["endCursor"]
    return results


def filter_by_period(assessments, start: date, end: date):
    """Geeft alleen assessments terug die zijn afgesloten binnen [start, end] (inclusief)."""
    filtered = []
    for a in assessments:
        closed = a.get("closedDate")
        if not closed:
            continue
        closed_date = date.fromisoformat(closed[:10])
        if start <= closed_date <= end:
            filtered.append(a)
    return filtered


def to_csv(assessments, output_path):
    if not assessments:
        print("Geen assessments gevonden voor deze periode.")
        return

    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([
            "Sequence ID", "Naam", "Effectiviteit",
            "Afgesloten op", "Aangemaakt op",
            "Control", "Control categorie",
            "Assets", "Uitvoerder",
        ])
        for a in assessments:
            controls = a.get("controls", [])
            control_names = ", ".join(c["name"] for c in controls)
            control_categories = ", ".join(c.get("category", "") for c in controls)
            assets = ", ".join(x["name"] for x in a.get("assets", []))
            executors = [
                x["user"]["email"]
                for x in a.get("assignments", [])
                if x["assignmentType"] == "EXECUTOR"
            ]
            writer.writerow([
                a["sequenceId"],
                a["name"],
                a["effectiveness"],
                a.get("closedDate", "")[:10] if a.get("closedDate") else "",
                a.get("createdDate", "")[:10] if a.get("createdDate") else "",
                control_names,
                control_categories,
                assets,
                ", ".join(executors),
            ])

    print(f"{len(assessments)} assessments geëxporteerd naar {output_path}")


if __name__ == "__main__":
    token = get_token()
    access_token = token["access_token"]

    all_assessments = fetch_all_assessments(access_token)

    # Februari 2025
    februari = filter_by_period(all_assessments, date(2025, 2, 1), date(2025, 2, 28))
    to_csv(februari, "effectiviteit_februari_2025.csv")

    # Q1 2025
    q1 = filter_by_period(all_assessments, date(2025, 1, 1), date(2025, 3, 31))
    to_csv(q1, "effectiviteit_q1_2025.csv")

Rapport van overdue controls

Haal alle tasks op die momenteel overdue zijn, inclusief de bijbehorende controls.

OVERDUE_QUERY = """
query OverdueTasks($first: Int, $after: String, $filter: ActivityFilter) {
  activities_paged(first: $first, after: $after, filter: $filter) {
    edges {
      node {
        sequenceId
        name
        expires
        controls { sequenceId name category }
        assets { name }
        assignments { assignmentType user { name email } }
      }
    }
    pageInfo { hasNextPage endCursor }
  }
}
"""


def fetch_overdue_tasks(access_token):
    results = []
    cursor = None
    while True:
        data = graphql(
            OVERDUE_QUERY,
            {"first": 50, "after": cursor, "filter": {"activityTypes": ["EXECUTE"], "status": ["OVERDUE"]}},
            access_token,
        )
        page = data["activities_paged"]
        results.extend(edge["node"] for edge in page["edges"])
        if not page["pageInfo"]["hasNextPage"]:
            break
        cursor = page["pageInfo"]["endCursor"]
    return results

Gecombineerd statusrapport

Dit voorbeeld combineert overdue tasks en effectiviteitsresultaten in één overzicht per control.

from collections import defaultdict


def build_control_summary(overdue_tasks, assessments):
    """
    Geeft een dict terug per control (sequenceId) met:
      - overdue_tasks: aantal momenteel overdue tasks
      - effective: aantal EFFECTIVE assessments
      - ineffective: aantal INEFFECTIVE assessments
      - undetermined: aantal UNDETERMINED assessments
    """
    summary = defaultdict(lambda: {
        "name": "",
        "category": "",
        "overdue_tasks": 0,
        "effective": 0,
        "ineffective": 0,
        "undetermined": 0,
    })

    for task in overdue_tasks:
        for control in task.get("controls", []):
            cid = control["sequenceId"]
            summary[cid]["name"] = control["name"]
            summary[cid]["category"] = control.get("category", "")
            summary[cid]["overdue_tasks"] += 1

    for a in assessments:
        for control in a.get("controls", []):
            cid = control["sequenceId"]
            summary[cid]["name"] = control["name"]
            summary[cid]["category"] = control.get("category", "")
            eff = a["effectiveness"].lower()
            if eff in ("effective", "ineffective", "undetermined"):
                summary[cid][eff] += 1

    return summary


def summary_to_csv(summary, output_path):
    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([
            "Control ID", "Naam", "Categorie",
            "Overdue tasks", "Effectief", "Ineffectief", "Onbepaald",
        ])
        for cid, row in sorted(summary.items()):
            writer.writerow([
                cid, row["name"], row["category"],
                row["overdue_tasks"], row["effective"],
                row["ineffective"], row["undetermined"],
            ])
    print(f"Overzicht geschreven naar {output_path}")


if __name__ == "__main__":
    token = get_token()
    access_token = token["access_token"]

    overdue = fetch_overdue_tasks(access_token)
    all_assessments = fetch_all_assessments(access_token)
    q1 = filter_by_period(all_assessments, date(2025, 1, 1), date(2025, 3, 31))

    summary = build_control_summary(overdue, q1)
    summary_to_csv(summary, "control_overzicht_q1_2025.csv")

Filteren op afdeling

Tidal Control heeft geen native "afdeling"- of "business unit"-veld op controls of tasks. Er zijn twee gangbare manieren om dit te modelleren:

Via assets: Als jouw organisatie controls koppelt aan assets die afdelingen of business units vertegenwoordigen, filter je na het ophalen op de naam van de asset:

afdeling = "Finance"
afdeling_controls = [
    c for c in controls
    if any(a["name"] == afdeling for a in c.get("assets", []))
]

Via control-attributen: Als jouw organisatie custom attributen gebruikt om controls van een afdeling te voorzien, filter je op het attributes-veld:

afdeling = "Finance"
afdeling_controls = [
    c for c in controls
    if any(
        attr["key"] == "department" and attr["value"] == afdeling
        for attr in c.get("attributes", [])
    )
]

Tip

Weet je niet hoe afdelingsinformatie in jouw Tidal-omgeving is ingericht? Neem contact op via support@tidalcontrol.com.