Running notebooks with the Deepnote API

curl -s "https://api.deepnote.com/v2/notebooks/$NOTEBOOK_ID" \
  -H "Authorization: Bearer $DEEPNOTE_TOKEN" | jq '.notebook.inputs'

[
  {
    "name": "region",
    "type": "input-select",
    "value": "EU",
    "options": ["EU", "US"]
  },
  { "name": "since", "type": "input-date", "value": "2026-01-01" }
]

curl -s -X POST https://api.deepnote.com/v2/runs \
  -H "Authorization: Bearer $DEEPNOTE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "notebookId": "'"$NOTEBOOK_ID"'",
    "inputs": { "region": "US", "since": "2026-10-01" }
  }'

{
  "runId": "4f6c…",
  "status": "pending",
  "createdAt": "2026-11-04T09:00:00.000Z"
}

curl -s "https://api.deepnote.com/v2/runs/$RUN_ID?snapshotDelivery=blocks" \
  -H "Authorization: Bearer $DEEPNOTE_TOKEN"

import os
import time

import requests

API = "https://api.deepnote.com/v2"
HEADERS = {"Authorization": f"Bearer {os.environ['DEEPNOTE_TOKEN']}"}
DONE = {"success", "error", "internal_error", "stopped"}
# (connect, read) timeouts, so a stalled request fails instead of hanging. The read timeout
# limits each wait for data, not the whole response, so `timeout` below is best effort: a
# request started just before the deadline can finish after it.
REQUEST_TIMEOUT = (5, 30)


def wait_for_retry(response: requests.Response, deadline: float, message: str) -> None:
    delay = int(response.headers["Retry-After"])
    if time.monotonic() + delay > deadline:
        raise TimeoutError(message)
    time.sleep(delay)


def poll_timeout(run_id: str, status: str | None, timeout: float) -> str:
    if status in DONE:
        return f"Run {run_id} ended with {status}, but its outputs weren't stored within {timeout} seconds"
    return f"Run {run_id} did not finish within {timeout} seconds"


def run_notebook(notebook_id: str, inputs: dict, timeout: float = 1800) -> list[dict]:
    deadline = time.monotonic() + timeout

    while True:
        if time.monotonic() > deadline:
            raise TimeoutError("Rate limited for longer than the timeout")
        response = requests.post(
            f"{API}/runs",
            headers=HEADERS,
            json={"notebookId": notebook_id, "inputs": inputs},
            timeout=REQUEST_TIMEOUT,
        )
        # A 429 means the run was not created, so starting it again is safe.
        if response.status_code == 429:
            wait_for_retry(response, deadline, "Rate limited for longer than the timeout")
            continue
        response.raise_for_status()
        break

    run_id = response.json()["runId"]

    status = None
    snapshot_checks = 0
    while True:
        if time.monotonic() > deadline:
            raise TimeoutError(poll_timeout(run_id, status, timeout))
        try:
            response = requests.get(
                f"{API}/runs/{run_id}",
                headers=HEADERS,
                params={"snapshotDelivery": "blocks"},
                timeout=REQUEST_TIMEOUT,
            )
        except (requests.ConnectionError, requests.Timeout):
            # Reading a run is safe to repeat, and the run keeps going meanwhile.
            time.sleep(max(0, min(5, deadline - time.monotonic())))
            continue
        if response.status_code == 429:
            wait_for_retry(response, deadline, poll_timeout(run_id, status, timeout))
            continue
        response.raise_for_status()
        result = response.json()["run"]
        status = result["status"]
        if status in DONE:
            # Outputs are stored a moment later; a notebook with nothing to run never gets them.
            if result["snapshotStatus"] != "pending" or snapshot_checks >= 6:
                break
            snapshot_checks += 1
        time.sleep(max(0, min(5, deadline - time.monotonic())))

    if status != "success":
        raise RuntimeError(result.get("error") or status)
    return result["snapshotBlocks"] or []


blocks = run_notebook(os.environ["NOTEBOOK_ID"], {"region": "US"})
for block in blocks:
    for output in block["outputs"]:
        print(block["type"], output.get("output_type"))

from IPython.display import display

display({"application/json": {"revenue": float(total), "orders": int(order_count)}}, raw=True)

# Every weekday at 07:00 New York time
curl -s -X POST "https://api.deepnote.com/v2/notebooks/$NOTEBOOK_ID/schedule" \
  -H "Authorization: Bearer $DEEPNOTE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{ "cron": "0 7 * * 1-5", "timezone": "America/New_York" }'