Use-case recipes

Guides

Use-case recipes

Step-by-step recipes for common collection and discovery tasks. Every sample is a self-contained, runnable Python script (only the requests library is required) — copy any one script and run it directly. For cross-cutting rules (pagination, dates, PascalCase responses) see the API conventions.

1 · List the seeded locations

Retrieve every source and destination location pre-configured in your trial tenant. Use the returned Id values as inputs to discovery calls and job policies.

import requests

BASE      = "https://api.dev.datatap.stream"
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

locations = get(f"/api/{TENANT_ID}/locations")
for loc in locations:
    print(loc["Id"], loc["Type"], loc.get("Name", "—"))

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"

curl -s \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/locations"

2 · List the seeded jobs

See all jobs provisioned by the trial setup. Each entry includes the job Id, Type, and current Enabled state — the Id is what you need to activate a job in recipe 5.

import requests

BASE      = "https://api.dev.datatap.stream"
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

jobs = get(f"/api/{TENANT_ID}/jobs")
for job in jobs:
    print(job["Id"], job["Name"], job["Type"], "enabled:", job.get("Enabled"))

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"

curl -s \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/jobs"

3 · List users from the email source

A single direct call against the seeded Exchange location returns the first page of available mailboxes — enough to inspect what the location sees before scoping a job. For a full paginated enumeration of every mailbox see recipe 6.

import requests

BASE      = "https://api.dev.datatap.stream"
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

locs = get(f"/api/{TENANT_ID}/locations")
exchange_loc = next(l["Id"] for l in locs if l["Type"] == "EXCHANGE")

# One-shot first page — enough to inspect what mailboxes the seeded location sees.
# For a full paginated enumeration of every mailbox see the "Discover all the mailboxes" recipe.
page = get(f"/api/{TENANT_ID}/discovery/{exchange_loc}/users",
           params={"PageSize": 100, "filter.type": "USERS_AND_SHAREDMAILBOXES"})
for user in page["Sources"]:
    print(user["Email"], "—", user["Name"], f"({user['Type']})")
print("continuation:", page["Options"].get("Marker", "none"))

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"
EXCHANGE_LOC="<exchange-location-id>"   # Id from the list-locations call above

curl -s \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/discovery/$EXCHANGE_LOC/users?PageSize=100&filter.type=USERS_AND_SHAREDMAILBOXES"

4 · List users from the OneDrive source

Same pattern against the seeded OneDrive location — returns drive owners whose drives can be targeted in a onedrive-backup policy. For the full paginated version see recipe 7.

import requests

BASE      = "https://api.dev.datatap.stream"
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

locs = get(f"/api/{TENANT_ID}/locations")
onedrive_loc = next(l["Id"] for l in locs if l["Type"] == "ONEDRIVE")

# One-shot first page of drive owners on the seeded OneDrive location.
# For full paginated enumeration see the "Discover all the OneDrive accounts" recipe.
page = get(f"/api/{TENANT_ID}/discovery/{onedrive_loc}/users",
           params={"PageSize": 100})
for user in page["Sources"]:
    print(user["Email"], "—", user["Name"])
print("continuation:", page["Options"].get("Marker", "none"))

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"
ONEDRIVE_LOC="<onedrive-location-id>"  # Id from the list-locations call above

curl -s \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/discovery/$ONEDRIVE_LOC/users?PageSize=100"

5 · Activate the seeded collection jobs

Seeded jobs are created with enabled: true and schedule.type: DISABLED — wired up but idle. Activating one is two steps: first PATCH the job to scope its policy to your target accounts (leave the schedule unchanged), then call POST /api/{tenant}/jobs/{jobId}/start to trigger an immediate run. The job must be in READY or FAILED state. Two scenarios are shown — the Enron public corpus sample and a placeholder organization policy.

import requests

BASE      = "https://api.dev.datatap.stream"
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

def post(path, json=None):
    r = requests.post(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, json=json, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

def patch(path, json=None):
    r = requests.patch(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, json=json, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

# Seeded jobs are created with enabled=true and schedule.type=DISABLED — ready but idle.
# Step 1: PATCH the job to scope its policy to the target users (leave schedule unchanged).
# Step 2: call /start to trigger an immediate run (STARTING → RUNNING → READY).
# The job must be in READY or FAILED state before it can be patched or started.
jobs = get(f"/api/{TENANT_ID}/jobs")

# ── Enron sample job ──────────────────────────────────────────────────────────────────────────────
enron_job = next(j for j in jobs if "enron" in j["Name"].lower())

patch(f"/api/{TENANT_ID}/jobs/{enron_job['Id']}", json={
    "id": enron_job["Id"],
    "policy": {
        "$type": "exchange-backup",
        "users": [
            {"item": {"id": "kenneth.lay",   "name": "Kenneth Lay",   "email": "kenneth.lay@enron.com",   "type": "User"}, "type": "INCLUDE"},
            {"item": {"id": "jeff.skilling", "name": "Jeff Skilling", "email": "jeff.skilling@enron.com", "type": "User"}, "type": "INCLUDE"}
        ]
    }
})
task = post(f"/api/{TENANT_ID}/jobs/{enron_job['Id']}/start")
print("Enron job started, taskId:", task["TaskId"])

# ── Organization job ──────────────────────────────────────────────────────────────────────────────
# Replace id/email with real values from the list-users-from-email-source recipe above.
org_job = next(j for j in jobs if "organization" in j["Name"].lower())

patch(f"/api/{TENANT_ID}/jobs/{org_job['Id']}", json={
    "id": org_job["Id"],
    "policy": {
        "$type": "exchange-backup",
        "users": [
            {"item": {"id": "user-1-id", "name": "Alice", "email": "alice@yourorg.com", "type": "User"}, "type": "INCLUDE"},
            {"item": {"id": "user-2-id", "name": "Bob",   "email": "bob@yourorg.com",   "type": "User"}, "type": "INCLUDE"}
        ]
    }
})
task = post(f"/api/{TENANT_ID}/jobs/{org_job['Id']}/start")
print("Org job started, taskId:", task["TaskId"])

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"
JOB_ID="<job-id>"   # Id from the list-jobs call above

# Step 1 — PATCH: scope the policy to your target users (schedule stays DISABLED):
curl -s -X PATCH \
  -H "X-API-Key: $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "id": "'"$JOB_ID"'",
    "policy": {
      "$type": "exchange-backup",
      "users": [
        {"item": {"id": "kenneth.lay",   "name": "Kenneth Lay",   "email": "kenneth.lay@enron.com",   "type": "User"}, "type": "INCLUDE"},
        {"item": {"id": "jeff.skilling", "name": "Jeff Skilling", "email": "jeff.skilling@enron.com", "type": "User"}, "type": "INCLUDE"}
      ]
    }
  }' \
  "https://api.dev.datatap.stream/api/$TENANT_ID/jobs/$JOB_ID"

# Step 2 — START: trigger an immediate run:
curl -s -X POST \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/jobs/$JOB_ID/start"

6 · Discover all the mailboxes

List every user and shared mailbox in your Exchange source, following the Options.Marker continuation token until it is empty.

import requests

BASE      = "https://api.dev.datatap.stream"  # your environment's API base URL
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"                       # sent as the X-API-Key header

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

locs = get(f"/api/{TENANT_ID}/locations")
EXCHANGE_LOC = next(l["Id"] for l in locs if l["Type"] == "EXCHANGE")

# Discovery's /users endpoint on a SOURCE location reads live Microsoft Graph. filter.type=
# USERS_AND_SHAREDMAILBOXES adds shared/resource mailboxes; omit it for user mailboxes only.
def discover_sources(location_id, include_shared=True):
    sources, marker = [], None
    while True:
        params = {"PageSize": 100}
        if include_shared:
            params["filter.type"] = "USERS_AND_SHAREDMAILBOXES"
        if marker:
            params["Marker"] = marker
        page = get(f"/api/{TENANT_ID}/discovery/{location_id}/users", params=params)
        sources.extend(page["Sources"])               # each: {Id, Name, Email, Type}
        marker = page["Options"].get("Marker")
        if not marker:
            break
    return sources

mailboxes = discover_sources(EXCHANGE_LOC)
print(f"{len(mailboxes)} mailboxes")
for m in mailboxes:
    print(m["Email"], "—", m["Name"], f"({m['Type']})")

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"
EXCHANGE_LOC="<exchange-location-id>"   # Id from GET /api/$TENANT_ID/locations

# First page (PageSize=100). Follow Options.Marker for subsequent pages.
curl -s \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/discovery/$EXCHANGE_LOC/users?PageSize=100&filter.type=USERS_AND_SHAREDMAILBOXES"

7 · Discover all the OneDrive accounts

The same paginated discovery call against the OneDrive source location returns every drive owner.

import requests

BASE      = "https://api.dev.datatap.stream"
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

locs = get(f"/api/{TENANT_ID}/locations")
ONEDRIVE_LOC = next(l["Id"] for l in locs if l["Type"] == "ONEDRIVE")

# Same paginated /users call against the OneDrive SOURCE location: returns drive owners.
# Shared mailboxes don't apply to OneDrive, so no filter.type needed.
def discover_sources(location_id):
    sources, marker = [], None
    while True:
        params = {"PageSize": 100}
        if marker:
            params["Marker"] = marker
        page = get(f"/api/{TENANT_ID}/discovery/{location_id}/users", params=params)
        sources.extend(page["Sources"])
        marker = page["Options"].get("Marker")
        if not marker:
            break
    return sources

accounts = discover_sources(ONEDRIVE_LOC)
print(f"{len(accounts)} OneDrive accounts")
for a in accounts:
    print(a["Email"], "—", a["Name"])

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"
ONEDRIVE_LOC="<onedrive-location-id>"  # Id from GET /api/$TENANT_ID/locations

# First page. Follow Options.Marker for subsequent pages.
curl -s \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/discovery/$ONEDRIVE_LOC/users?PageSize=100"

8 · Collect emails & notes from two user accounts

Create a BACKUP job whose policy includes only the two chosen mailboxes. An Exchange collection collects their mail and notes; scope further with folder includes (IndexFromPaths) if you need only specific folders. Indexing for search is a premium feature, so the job runs with indexFiles: false.

import requests, time

BASE      = "https://api.dev.datatap.stream"
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

def post(path, json=None):
    r = requests.post(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, json=json, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

locs = get(f"/api/{TENANT_ID}/locations")
EXCHANGE_LOC = next(l["Id"] for l in locs if l["Type"] == "EXCHANGE")
BLOB_LOC     = next(l["Id"] for l in locs if l["Type"] == "BLOB")

# Scope a collection to two specific mailboxes via the policy's Users include-list. An Exchange
# collection collects each user's mail (all folders) and notes. (It also captures calendar items; to
# restrict to specific folders, add IndexFromPaths includes.)
def discover_sources(location_id, include_shared=True):
    sources, marker = [], None
    while True:
        params = {"PageSize": 100}
        if include_shared:
            params["filter.type"] = "USERS_AND_SHAREDMAILBOXES"
        if marker:
            params["Marker"] = marker
        page = get(f"/api/{TENANT_ID}/discovery/{location_id}/users", params=params)
        sources.extend(page["Sources"])
        marker = page["Options"].get("Marker")
        if not marker:
            break
    return sources

mboxes = {m["Email"].lower(): m for m in discover_sources(EXCHANGE_LOC)}
picked = [mboxes["alice@contoso.com"], mboxes["bob@contoso.com"]]

job = post(f"/api/{TENANT_ID}/jobs", json={
    "name": "Emails & notes — Alice + Bob",
    "description": "Collection of two mailboxes",
    "priority": "Medium",
    "type": "BACKUP",
    "schedule": {"type": "NOW"},
    "sourceId": EXCHANGE_LOC,
    "destinationId": BLOB_LOC,
    "indexFiles": False,                               # indexing is a premium (paid) feature
    "owner": "me@contoso.com",
    "policy": {
        "$type": "exchange-backup",                   # polymorphic discriminator
        "users": [                                     # INCLUDE only these two mailboxes
            {"item": {"id": u["Id"], "name": u["Name"], "email": u["Email"], "type": u["Type"]},
             "type": "INCLUDE"}
            for u in picked
        ]
    }
})
JOB_ID = job["Id"]

task = post(f"/api/{TENANT_ID}/jobs/{JOB_ID}/start")
TASK_ID = task["TaskId"]
while True:                                            # STARTING -> RUNNING -> READY | FAILED
    st = get(f"/api/{TENANT_ID}/jobs/{JOB_ID}/status/{TASK_ID}")
    print("state:", st["State"])
    if st["State"] in ("READY", "FAILED"):
        break
    time.sleep(10)

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"
EXCHANGE_LOC="<exchange-location-id>"
BLOB_LOC="<blob-location-id>"

# 1 · Create the BACKUP job scoped to two mailboxes:
JOB=$(curl -s -X POST \
  -H "X-API-Key: $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Emails & notes — Alice + Bob",
    "priority": "Medium",
    "type": "BACKUP",
    "schedule": {"type": "NOW"},
    "sourceId": "'"$EXCHANGE_LOC"'",
    "destinationId": "'"$BLOB_LOC"'",
    "indexFiles": false,
    "owner": "me@contoso.com",
    "policy": {
      "$type": "exchange-backup",
      "users": [
        {"item": {"id": "<alice-id>", "name": "Alice", "email": "alice@contoso.com", "type": "User"}, "type": "INCLUDE"},
        {"item": {"id": "<bob-id>",   "name": "Bob",   "email": "bob@contoso.com",   "type": "User"}, "type": "INCLUDE"}
      ]
    }
  }' \
  "https://api.dev.datatap.stream/api/$TENANT_ID/jobs")
JOB_ID=$(echo $JOB | python3 -c "import sys,json; print(json.load(sys.stdin)['Id'])")

# 2 · Start it and poll status:
TASK=$(curl -s -X POST \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/jobs/$JOB_ID/start")
TASK_ID=$(echo $TASK | python3 -c "import sys,json; print(json.load(sys.stdin)['TaskId'])")

curl -s \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/jobs/$JOB_ID/status/$TASK_ID"

9 · List collected Sent Items for a date range

Browse the collected mailbox on the BLOB location: find its Sent Items folder, then page its messages between two timestamps. The mailbox must have been collected first (recipe 8). Each result carries the ItemId + FragmentId you need to download a single message as .eml.

import requests

BASE      = "https://api.dev.datatap.stream"
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

locs = get(f"/api/{TENANT_ID}/locations")
BLOB_LOC = next(l["Id"] for l in locs if l["Type"] == "BLOB")

# "Collected" data is browsed on the BLOB collection location (not the live source). First list the
# mailbox's folders to find Sent Items, then list its MessageItems within the date range.
SOURCE = "alice@contoso.com"                           # the collected mailbox (its owner id/email)

folders = get(f"/api/{TENANT_ID}/discovery/{BLOB_LOC}/source/{SOURCE}/items/-",
              params={"types": "MailboxFolder", "PageSize": 100})["Items"]
sent = next(f for f in folders
            if f.get("Discriminator") == "SentMailboxFolder" or f["Name"] == "Sent Items")

# Listing MessageItems requires a full-timestamp date range. Each item carries an ItemId + FragmentId
# (use them with the /download endpoint to pull a single message as .eml).
def list_messages(folder_id, from_date, to_date):
    items, marker = [], None
    while True:
        params = {"types": "MessageItem", "fromDate": from_date, "toDate": to_date, "PageSize": 100}
        if marker:
            params["Marker"] = marker
        page = get(f"/api/{TENANT_ID}/discovery/{BLOB_LOC}/source/{SOURCE}/items/{folder_id}",
                   params=params)
        items.extend(page["Items"])
        marker = page["Options"].get("Marker")
        if not marker:
            break
    return items

sent_items = list_messages(sent["ItemId"], "2024-01-01T00:00:00Z", "2024-03-31T23:59:59Z")
print(f"{len(sent_items)} sent items in range")
for it in sent_items:
    print(it["CreatedDate"], it["Name"], "fragment:", it["FragmentId"])

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"
BLOB_LOC="<blob-location-id>"
SOURCE="alice@contoso.com"

# 1 · List folders to find the Sent Items folder:
curl -s \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/discovery/$BLOB_LOC/source/$SOURCE/items/-?types=MailboxFolder&PageSize=100"

# 2 · List messages in Sent Items (replace FOLDER_ID with the ItemId from step 1):
FOLDER_ID="<sent-items-folder-item-id>"
curl -s \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/discovery/$BLOB_LOC/source/$SOURCE/items/$FOLDER_ID?types=MessageItem&fromDate=2024-01-01T00%3A00%3A00Z&toDate=2024-03-31T23%3A59%3A59Z&PageSize=100"

10 · Collect all files from a user OneDrive

Collect one user's entire drive with a OneDrive collection scoped to that drive owner.

import requests

BASE      = "https://api.dev.datatap.stream"
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

def post(path, json=None):
    r = requests.post(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, json=json, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

locs = get(f"/api/{TENANT_ID}/locations")
ONEDRIVE_LOC = next(l["Id"] for l in locs if l["Type"] == "ONEDRIVE")
BLOB_LOC     = next(l["Id"] for l in locs if l["Type"] == "BLOB")

def discover_sources(location_id):
    sources, marker = [], None
    while True:
        params = {"PageSize": 100}
        if marker:
            params["Marker"] = marker
        page = get(f"/api/{TENANT_ID}/discovery/{location_id}/users", params=params)
        sources.extend(page["Sources"])
        marker = page["Options"].get("Marker")
        if not marker:
            break
    return sources

# Collect a single user's entire OneDrive: a OneDrive collection scoped to one drive owner.
owner = next(a for a in discover_sources(ONEDRIVE_LOC)
             if a["Email"].lower() == "alice@contoso.com")

job = post(f"/api/{TENANT_ID}/jobs", json={
    "name": "OneDrive collection — Alice",
    "description": "Full drive collection",
    "priority": "Medium",
    "type": "BACKUP",
    "schedule": {"type": "NOW"},
    "sourceId": ONEDRIVE_LOC,
    "destinationId": BLOB_LOC,
    "indexFiles": False,
    "owner": "me@contoso.com",
    "policy": {
        "$type": "onedrive-backup",
        "users": [{"item": {"id": owner["Id"], "name": owner["Name"],
                            "email": owner["Email"], "type": owner["Type"]},
                   "type": "INCLUDE"}]
    }
})
post(f"/api/{TENANT_ID}/jobs/{job['Id']}/start")

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"
ONEDRIVE_LOC="<onedrive-location-id>"
BLOB_LOC="<blob-location-id>"
ALICE_ID="<alice-user-id>"   # Id from GET /discovery/$ONEDRIVE_LOC/users

# Create the BACKUP job scoped to Alice's drive:
JOB=$(curl -s -X POST \
  -H "X-API-Key: $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "OneDrive collection — Alice",
    "priority": "Medium",
    "type": "BACKUP",
    "schedule": {"type": "NOW"},
    "sourceId": "'"$ONEDRIVE_LOC"'",
    "destinationId": "'"$BLOB_LOC"'",
    "indexFiles": false,
    "owner": "me@contoso.com",
    "policy": {
      "$type": "onedrive-backup",
      "users": [
        {"item": {"id": "'"$ALICE_ID"'", "name": "Alice", "email": "alice@contoso.com", "type": "OneDriveUser"}, "type": "INCLUDE"}
      ]
    }
  }' \
  "https://api.dev.datatap.stream/api/$TENANT_ID/jobs")
JOB_ID=$(echo $JOB | python3 -c "import sys,json; print(json.load(sys.stdin)['Id'])")

curl -s -X POST \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/jobs/$JOB_ID/start"

11 · List & download the PDF files from a user drive

Walk the collected drive and keep the .pdf files (filter on the file name — there is no server-side extension filter). The item /download endpoint serves emails only, so to retrieve the actual file bytes run a RESTORE job scoped to PDFs, which writes them back into the user's OneDrive.

import requests

BASE      = "https://api.dev.datatap.stream"
TENANT_ID = "<your-tenant-id>"
API_KEY   = "<your-tenant-api-key>"

def get(path, params=None):
    r = requests.get(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, params=params, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

def post(path, json=None):
    r = requests.post(f"{BASE}{path}", headers={"X-API-Key": API_KEY}, json=json, timeout=60)
    r.raise_for_status()
    return r.json() if r.content else None

locs = get(f"/api/{TENANT_ID}/locations")
BLOB_LOC     = next(l["Id"] for l in locs if l["Type"] == "BLOB")
ONEDRIVE_LOC = next(l["Id"] for l in locs if l["Type"] == "ONEDRIVE")

SOURCE = "alice@contoso.com"

# 1 · LIST — walk the collected drive (BLOB location) and keep the .pdf files. Browsing returns
#     OneDriveFile / OneDriveFolder nodes; recurse folders by their ItemId. There is no server-side
#     extension filter, so match on the file Name.
def list_children(node_id, types):
    items, marker = [], None
    while True:
        params = {"types": types, "PageSize": 100}
        if marker:
            params["Marker"] = marker
        page = get(f"/api/{TENANT_ID}/discovery/{BLOB_LOC}/source/{SOURCE}/items/{node_id}",
                   params=params)
        items.extend(page["Items"])
        marker = page["Options"].get("Marker")
        if not marker:
            break
    return items

def walk_pdfs(node_id="-"):
    pdfs = [f for f in list_children(node_id, "OneDriveFile")
            if f["Name"].lower().endswith(".pdf")]
    for folder in list_children(node_id, "OneDriveFolder"):
        pdfs += walk_pdfs(folder["ItemId"])
    return pdfs

pdfs = walk_pdfs()
print(f"{len(pdfs)} PDFs:", [p["ReadablePath"] for p in pdfs])

# 2 · DOWNLOAD — the /download endpoint currently serves emails only (dataType=email). To get the
#     actual PDF bytes back, run a RESTORE job scoped to PDFs; it writes them into the user's drive.
restore = post(f"/api/{TENANT_ID}/jobs", json={
    "name": "Restore PDFs — Alice",
    "description": "Restore .pdf files to OneDrive",
    "priority": "High",
    "type": "RESTORE",
    "schedule": {"type": "NOW"},
    "sourceId": BLOB_LOC,                              # restore FROM the collection
    "destinationId": ONEDRIVE_LOC,                     # …back INTO OneDrive
    "indexFiles": False,
    "owner": "me@contoso.com",
    "policy": {
        "$type": "onedrive-restore",
        "users": [{"item": {"id": "<alice-id>", "name": "Alice",
                            "email": "alice@contoso.com", "type": "OneDriveUser"}, "type": "INCLUDE"}],
        "extensions": [{"item": "PDF", "type": "INCLUDE"}]   # only .pdf files
    }
})
post(f"/api/{TENANT_ID}/jobs/{restore['Id']}/start")

API_KEY="<your-tenant-api-key>"
TENANT_ID="<your-tenant-id>"
BLOB_LOC="<blob-location-id>"
ONEDRIVE_LOC="<onedrive-location-id>"
SOURCE="alice@contoso.com"
ALICE_ID="<alice-user-id>"

# 1 · List the root of Alice's collected drive (repeat with ItemId to recurse into folders):
curl -s \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/discovery/$BLOB_LOC/source/$SOURCE/items/-?types=OneDriveFile&PageSize=100"

# 2 · Create a RESTORE job scoped to .pdf files only:
JOB=$(curl -s -X POST \
  -H "X-API-Key: $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Restore PDFs — Alice",
    "priority": "High",
    "type": "RESTORE",
    "schedule": {"type": "NOW"},
    "sourceId": "'"$BLOB_LOC"'",
    "destinationId": "'"$ONEDRIVE_LOC"'",
    "indexFiles": false,
    "owner": "me@contoso.com",
    "policy": {
      "$type": "onedrive-restore",
      "users": [
        {"item": {"id": "'"$ALICE_ID"'", "name": "Alice", "email": "alice@contoso.com", "type": "OneDriveUser"}, "type": "INCLUDE"}
      ],
      "extensions": [{"item": "PDF", "type": "INCLUDE"}]
    }
  }' \
  "https://api.dev.datatap.stream/api/$TENANT_ID/jobs")
JOB_ID=$(echo $JOB | python3 -c "import sys,json; print(json.load(sys.stdin)['Id'])")

curl -s -X POST \
  -H "X-API-Key: $API_KEY" \
  "https://api.dev.datatap.stream/api/$TENANT_ID/jobs/$JOB_ID/start"