API Reference
Reporting use cases
These examples show how to build the operational reports that compliance teams commonly need: control effectiveness per period, overdue task tracking, and combined status overviews.
The API does not support server-side date range filters. All period filtering (e.g. "February only", "Q1") is done client-side by filtering on the closedDate field after fetching data.
The scripts on this page use get_token() from the Authentication page. Copy that function into your script, or import it from a shared module.
Control effectiveness report (by period)
Assessments are the source of truth for effectiveness. Each assessment has a closedDate (when the task was completed) and an effectiveness value (EFFECTIVE, INEFFECTIVE, or UNDETERMINED).
To report on a specific month or quarter, fetch all assessments and filter by closedDate in your 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):
"""Keep only assessments closed within [start, end] (inclusive)."""
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("No assessments found for this period.")
return
with open(output_path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow([
"Sequence ID", "Name", "Effectiveness",
"Closed Date", "Created Date",
"Control", "Control Category",
"Assets", "Executor",
])
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"Exported {len(assessments)} assessments to {output_path}")
if __name__ == "__main__":
token = get_token()
access_token = token["access_token"]
all_assessments = fetch_all_assessments(access_token)
# February 2025
february = filter_by_period(all_assessments, date(2025, 2, 1), date(2025, 2, 28))
to_csv(february, "effectiveness_february_2025.csv")
# Q1 2025
q1 = filter_by_period(all_assessments, date(2025, 1, 1), date(2025, 3, 31))
to_csv(q1, "effectiveness_q1_2025.csv")
Overdue controls report
Fetch all tasks that are currently overdue, with the controls they belong to.
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
Combined status report
This example combines both overdue tasks and effectiveness results into a single per-control summary.
from collections import defaultdict
def build_control_summary(overdue_tasks, assessments):
"""
Returns a dict keyed by control sequenceId with:
- overdue_tasks: count of currently overdue tasks
- effective: count of EFFECTIVE assessments
- ineffective: count of INEFFECTIVE assessments
- undetermined: count of 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", "Name", "Category",
"Overdue Tasks", "Effective", "Ineffective", "Undetermined",
])
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"Summary written to {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_summary_q1_2025.csv")
Filtering by department
Tidal Control does not have a native "department" or "business unit" field on controls or tasks. There are two common ways to model this:
Using assets: If your organisation links controls to assets that represent departments or business units, filter by asset name after fetching:
department = "Finance"
dept_controls = [
c for c in controls
if any(a["name"] == department for a in c.get("assets", []))
]
Using control attributes: If your organisation uses custom attributes to tag controls with a department, filter on the attributes field:
department = "Finance"
dept_controls = [
c for c in controls
if any(
attr["key"] == "department" and attr["value"] == department
for attr in c.get("attributes", [])
)
]
If neither of these fits your setup, contact support@tidalcontrol.com to discuss how your organisation has structured department information in Tidal.
- Previous
- Exporting tasks