API-referentie
Rapportage use cases
Deze voorbeelden laten zien hoe je de operationele rapporten bouwt die compliance teams vaak nodig hebben: maatregeleffectiviteit per periode, overzichten van verlopen taken, en gecombineerde statusrapporten.
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.
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.
Maatregeleffectiviteitsrapport (per periode)
Beoordelingen zijn de bron voor effectiviteit. Elke beoordeling 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 beoordelingen 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 beoordelingen 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 beoordelingen 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",
"Maatregel", "Maatregelcategorie",
"Middelen", "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)} beoordelingen 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 verlopen maatregelen
Haal alle taken op die momenteel verlopen zijn, inclusief de bijbehorende maatregelen.
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 verlopen taken en effectiviteitsresultaten in één overzicht per maatregel.
from collections import defaultdict
def build_control_summary(overdue_tasks, assessments):
"""
Geeft een dict terug per maatregel (sequenceId) met:
- overdue_tasks: aantal momenteel verlopen taken
- effective: aantal EFFECTIVE beoordelingen
- ineffective: aantal INEFFECTIVE beoordelingen
- undetermined: aantal UNDETERMINED beoordelingen
"""
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([
"Maatregel-ID", "Naam", "Categorie",
"Verlopen taken", "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 maatregelen of taken. Er zijn twee gangbare manieren om dit te modelleren:
Via middelen: Als jouw organisatie maatregelen koppelt aan middelen die afdelingen of business units vertegenwoordigen, filter je na het ophalen op de naam van het middel:
afdeling = "Finance"
afdeling_controls = [
c for c in controls
if any(a["name"] == afdeling for a in c.get("assets", []))
]
Via maatregelattributen: Als jouw organisatie custom attributen gebruikt om maatregelen 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", [])
)
]
Weet je niet hoe afdelingsinformatie in jouw Tidal-omgeving is ingericht? Neem contact op via support@tidalcontrol.com.
- Vorige
- Taken exporteren
- Volgende
- Schema-introspectie