Usage Guide

Overview

DataGuard is ADEK's data classification and governance API. It identifies data points in natural-language queries, classifies them against ADEK's data governance framework (Open, Confidential, Sensitive, Secret), and answers policy questions grounded in uploaded governance documents.

Three main capabilities are exposed via the API:

• Classify : identify and classify data points from free-text queries.
• Policy Search : ask governance and policy questions in plain English.
• Bulk Classify : submit batches of data points for background classification.

Prerequisites

Install the required Python packages:

pip install requests msal

• requests : HTTP client for calling the API.
• msal : Microsoft Authentication Library. Only needed in production (Azure AD).

Authentication

DataGuard supports two authentication modes depending on the environment.

Staging: API Key

In staging, pass your API key in the X-API-Key header:

import requests

BASE_URL = "https://your-container-app.azurecontainerapps.io"
API_KEY  = "your-api-key"

headers = {
    "Content-Type": "application/json",
    "X-API-Key": API_KEY,
}

Production: Azure AD (Easy Auth)

In production, API keys are disabled. Authenticate with Azure AD using MSAL to obtain an access token, then call the API through the Easy Auth layer.

import msal
import requests

# Azure AD app registration details (get from your administrator)
TENANT_ID     = "your-tenant-id"
CLIENT_ID     = "your-client-app-id"          # YOUR app registration
CLIENT_SECRET = "your-client-secret"
API_APP_ID    = "your-dataguard-api-app-id"   # DataGuard's app registration

BASE_URL = "https://your-container-app.azurecontainerapps.io"
SCOPES   = [f"api://{API_APP_ID}/.default"]

# Confidential client (service-to-service / daemon flow)
app = msal.ConfidentialClientApplication(
    CLIENT_ID,
    authority=f"https://login.microsoftonline.com/{TENANT_ID}",
    client_credential=CLIENT_SECRET,
)

def get_access_token():
    result = app.acquire_token_for_client(scopes=SCOPES)
    if "access_token" in result:
        return result["access_token"]
    raise RuntimeError(f"Token error: {result.get('error_description')}")

token = get_access_token()

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {token}",
}

Classify

Classify data points mentioned in a natural-language query.

Request

Example

response = requests.post(
    f"{BASE_URL}/api/v1/classify",
    headers=headers,
    json={"query": "employee salary and national ID number"},
)
response.raise_for_status()
result = response.json()

print(f"Status: {result['pipeline_status']}")
print(f"Composite level: {result['composite_level']}")

for item in result["results"]:
    print(f"  {item['data_point']}: {item['classification_level']}")
    print(f"    Confidence: {item['confidence']} : {item['confidence_reason']}")
    print(f"    Summary: {item['summary']}")

Response Fields

Each Result Item

Policy Search

Ask governance and policy questions. Returns answers grounded in ADEK's uploaded policy documents with citations.

Request

Example

response = requests.post(
    f"{BASE_URL}/api/v1/policy-search",
    headers=headers,
    json={"query": "What is the retention period for financial records?"},
)
response.raise_for_status()
result = response.json()

print(f"Answer: {result['answer']}")

for cite in result.get("citations", []):
    print(f"  Source: {cite['document_name']}")

Response Fields

Bulk Classification

Submit a batch of data points for background classification. This is a three-step process: submit, poll, then download.

items = [
    {"data_point": "Employee Salary", "description": "Monthly gross salary"},
    {"data_point": "National ID Number"},
    {"data_point": "Office Phone Number"},
]

response = requests.post(
    f"{BASE_URL}/api/v1/admin/bulk-classify",
    headers=headers,
    json={"items": items},
)
response.raise_for_status()
job = response.json()
job_id = job["job_id"]
print(f"Submitted: {job_id} ({job['total_items']} items)")

import time

while True:
    resp = requests.get(
        f"{BASE_URL}/api/v1/jobs/{job_id}",
        headers=headers,
    )
    status = resp.json()

    print(f"  Status: {status['status']} : {status.get('message', '')}")

    if status["status"] in ("completed", "failed"):
        break

    time.sleep(5)  # poll every 5 seconds

download = requests.get(
    f"{BASE_URL}/api/v1/admin/bulk-classify/{job_id}/results",
    headers=headers,
).json()

if download.get("download_url"):
    results = requests.get(download["download_url"]).json()

    print(f"Classified: {results['classified_count']}/{results['total_items']}")
    for r in results["results"]:
        print(f"  {r['data_point']}: {r['classification_level']} ({r['confidence']})")

Result Item Fields

Job Status

Check the status of any background job (bulk classification, index rebuilds).

Response Fields

Error Handling

All endpoints return standard HTTP error codes with a JSON body containing a detail field.

try:
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    data = response.json()
except requests.exceptions.HTTPError as e:
    error = e.response.json().get("detail", str(e))
    print(f"HTTP {e.response.status_code}: {error}")

Common Error Codes

Roles & Permissions

Azure AD app roles control access to different endpoint tiers in production.

Quick Reference

Environment Modes