feat(streamlit): add Deepnote app helpers

.github/workflows/ci.yml

@@ -333,6 +333,10 @@ jobs:
       - name: Install project
         run: poetry install --no-interaction --only-root
 
+      - name: Install Streamlit for widget integration tests
+        if: matrix.python-version == '3.12'
+        run: poetry install --no-interaction --extras server
+
       - name: Run unit tests
         env:
           TOOLKIT_VERSION: ${{ steps.version.outputs.VERSION }}

README.md

@@ -22,6 +22,8 @@ It starts and manages Jupyter, Streamlit, and LSP servers, and provides runtime
 - Native **Deepnote component library** including beautiful `DataFrame` rendering and interactive inputs
 - **Python kernel with curated set of libraries preinstalled**, allowing you to focus on work instead of fighting with Python dependencies
 - Run multiple **interactive applications built with Streamlit**
+- Build custom Streamlit interfaces over local Deepnote files and hosted runs with
+  [per-viewer authentication](docs/streamlit-apps.md)
 - Language Server Protocol integration for code completion and intelligence
 - Git integration with SSH/HTTPS authentication
 

deepnote_toolkit/notebooks/__init__.py

@@ -0,0 +1,28 @@
+"""Read `.deepnote` files and run notebooks, independent of any UI framework.
+
+Only the names in __all__ are supported public API. Wire schemas and API clients
+are implementation details.
+"""
+
+from .cloud_runner import DeepnoteCloudRunner
+from .credentials import ApiCredentials, CredentialsProvider
+from .document import DeepnoteDocument
+from .local_runner import DeepnoteLocalRunner
+from .models import DeepnoteDataframe, InputBlock, NotebookOutput, RunnerInfo
+from .run_result import RunResult
+from .runner import Runner, RunnerError
+
+__all__ = [
+    "ApiCredentials",
+    "CredentialsProvider",
+    "DeepnoteCloudRunner",
+    "DeepnoteDataframe",
+    "DeepnoteDocument",
+    "DeepnoteLocalRunner",
+    "InputBlock",
+    "NotebookOutput",
+    "RunResult",
+    "Runner",
+    "RunnerError",
+    "RunnerInfo",
+]

deepnote_toolkit/notebooks/_schemas.py

@@ -0,0 +1,62 @@
+"""Consumed fields of the v2 API contracts (contracts/runs.ts and notebooks.ts).
+
+Extra fields are ignored.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field, StrictBool, StrictFloat, StrictInt, StrictStr
+
+from .api_types import RunStatus, SnapshotStatus
+
+
+class ApiInput(BaseModel):
+    name: StrictStr
+    type: StrictStr
+    value: StrictStr | StrictBool | list[StrictStr] | None = None
+    label: StrictStr | None = None
+    options: list[StrictStr] = Field(default_factory=list)
+    multiple: StrictBool = False
+    min: StrictInt | StrictFloat | None = None
+    max: StrictInt | StrictFloat | None = None
+    step: StrictInt | StrictFloat | None = None
+
+
+class ApiNotebook(BaseModel):
+    name: StrictStr = "Untitled notebook"
+    inputs: list[ApiInput] = Field(default_factory=list)
+
+
+class NotebookResponse(BaseModel):
+    notebook: ApiNotebook
+
+
+class CreateRunResponse(BaseModel):
+    """The run identity and execution status returned by POST /v2/runs."""
+
+    run_id: StrictStr = Field(alias="runId", min_length=1)
+    status: RunStatus
+
+
+class ApiRun(CreateRunResponse):
+    """GET run details, including the required snapshot lifecycle status."""
+
+    snapshot_status: SnapshotStatus = Field(alias="snapshotStatus")
+    snapshot_blocks: list[dict[str, Any]] | None = Field(
+        default=None, alias="snapshotBlocks"
+    )
+    error: StrictStr | None = None
+
+
+class GetRunResponse(BaseModel):
+    """The response envelope returned when fetching an existing run."""
+
+    run: ApiRun
+
+
+class ViewerTokenResponse(BaseModel):
+    token: StrictStr = Field(min_length=1)
+    api_origin: StrictStr = Field(alias="apiOrigin")
+    expires_at_seconds: StrictInt | StrictFloat = Field(alias="expiresAtSeconds")

deepnote_toolkit/notebooks/api_client.py

@@ -0,0 +1,221 @@
+"""The Deepnote public API operations a runner needs."""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from typing import Any, TypeVar, cast
+from urllib.parse import quote
+
+import requests
+from pydantic import BaseModel, ValidationError
+
+from ._schemas import (
+    ApiInput,
+    ApiRun,
+    CreateRunResponse,
+    GetRunResponse,
+    NotebookResponse,
+)
+from .api_types import (
+    INPUT_BLOCK_TYPES,
+    TERMINAL_RUN_STATUSES,
+    InputBlockType,
+    InputValue,
+    RunStatus,
+    SnapshotStatus,
+    StorageMode,
+)
+from .credentials import CredentialsProvider
+from .models import InputBlock, NotebookOutput
+from .runner import RunnerError
+from .transport import request_json
+from .wire import decode_block_outputs
+
+Schema = TypeVar("Schema", bound=BaseModel)
+
+
+@dataclass(frozen=True)
+class CloudNotebook:
+    """A notebook's name and input blocks."""
+
+    name: str
+    inputs: tuple[InputBlock, ...]
+
+
+@dataclass(frozen=True)
+class CloudRun:
+    """The state of one run. `outputs` is None until the run's snapshot is stored."""
+
+    run_id: str
+    status: RunStatus
+    snapshot_status: SnapshotStatus | None
+    outputs: tuple[NotebookOutput, ...] | None
+    error: str | None
+
+    @property
+    def is_finished(self) -> bool:
+        """Whether the run has reached a final status."""
+
+        return self.status in TERMINAL_RUN_STATUSES
+
+
+class DeepnoteApiClient:
+    """Sends requests to the Deepnote public API and validates what comes back."""
+
+    def __init__(
+        self,
+        credentials: CredentialsProvider,
+        *,
+        session: requests.Session | None = None,
+        request_timeout: float = 30,
+        clock: Callable[[], float] = time.monotonic,
+    ):
+        self._credentials = credentials
+        self._clock = clock
+        self._session = session if session is not None else requests.Session()
+        self._request_timeout = request_timeout
+
+    def get_notebook(self, notebook_id: str) -> CloudNotebook:
+        """Read a notebook's name and input blocks."""
+
+        payload = self._request("GET", f"/v2/notebooks/{quote(notebook_id, safe='')}")
+        notebook = _validate(NotebookResponse, payload, "notebook").notebook
+        return CloudNotebook(
+            name=notebook.name,
+            inputs=tuple(
+                _input_block(value)
+                for value in notebook.inputs
+                if value.type in INPUT_BLOCK_TYPES and value.name
+            ),
+        )
+
+    def create_run(
+        self,
+        notebook_id: str,
+        inputs: Mapping[str, Any],
+        *,
+        storage_mode: StorageMode | None = None,
+        timeout: float | None = None,
+    ) -> CloudRun:
+        """Start a detached run of the whole notebook."""
+
+        body: dict[str, Any] = {
+            "notebookId": notebook_id,
+            "detached": True,
+            "inputs": {
+                name: _encode_input(name, value) for name, value in inputs.items()
+            },
+        }
+        if storage_mode is not None:
+            body["detachedRunStorageMode"] = storage_mode
+        payload = self._request("POST", "/v2/runs", body, timeout=timeout)
+        run = _validate(CreateRunResponse, payload, "run")
+        return CloudRun(
+            run_id=run.run_id,
+            status=run.status,
+            snapshot_status=None,
+            outputs=None,
+            error=None,
+        )
+
+    def get_run(self, run_id: str, *, timeout: float | None = None) -> CloudRun:
+        """Read a run with the outputs of the notebook it executed."""
+
+        # The blocks delivery holds the executed notebook alone, not the whole project.
+        payload = self._request(
+            "GET",
+            f"/v2/runs/{quote(run_id, safe='')}?snapshotDelivery=blocks",
+            timeout=timeout,
+        )
+        return _cloud_run(_validate(GetRunResponse, payload, "run").run)
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        body: Mapping[str, Any] | None = None,
+        *,
+        timeout: float | None = None,
+    ) -> Mapping[str, Any]:
+        budget = (
+            self._request_timeout
+            if timeout is None
+            else min(timeout, self._request_timeout)
+        )
+        if budget <= 0:
+            raise RunnerError("API request deadline expired", transient=True)
+        deadline = self._clock() + budget
+        credentials = self._credentials(timeout=budget)
+        remaining = deadline - self._clock()
+        if remaining <= 0:
+            raise RunnerError(
+                "API credentials exhausted the request timeout", transient=True
+            )
+        return request_json(
+            self._session,
+            method,
+            f"{credentials.api_origin}{path}",
+            headers={"Authorization": f"Bearer {credentials.token}"},
+            body=body,
+            timeout=remaining,
+        )
+
+
+def _validate(schema: type[Schema], payload: Mapping[str, Any], what: str) -> Schema:
+    """Validate an API response, reporting schema failures as `RunnerError`."""
+
+    try:
+        return schema(**payload)
+    except ValidationError as error:
+        raise RunnerError(
+            f"Deepnote API returned an invalid {what} response"
+        ) from error
+
+
+def _encode_input(name: str, value: Any) -> InputValue:
+    """Convert a value to the form the runs API accepts, or raise `ValueError`."""
+
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (list, tuple)):
+        return [str(item) for item in value]
+    if value is None or isinstance(value, (Mapping, set, frozenset, bytes)):
+        raise ValueError(
+            f'Input "{name}" has a {type(value).__name__} value. '
+            "Pass text, a number, a boolean or a list of texts."
+        )
+    return str(value)
+
+
+def _input_block(value: ApiInput) -> InputBlock:
+    """Convert validated API input metadata to a notebook input block."""
+
+    return InputBlock(
+        variable_name=value.name,
+        type=cast(InputBlockType, value.type),
+        value=value.value,
+        label=value.label,
+        options=tuple(value.options),
+        multiple=value.multiple,
+        min=value.min,
+        max=value.max,
+        step=value.step,
+    )
+
+
+def _cloud_run(run: ApiRun) -> CloudRun:
+    """Convert a validated run and any available snapshot blocks to runner data."""
+
+    return CloudRun(
+        run_id=run.run_id,
+        status=run.status,
+        snapshot_status=run.snapshot_status,
+        outputs=(
+            decode_block_outputs(run.snapshot_blocks, id_key="id")
+            if run.snapshot_blocks is not None
+            else None
+        ),
+        error=run.error,
+    )

deepnote_toolkit/notebooks/api_types.py

@@ -0,0 +1,27 @@
+"""Value types of the Deepnote public API."""
+
+from __future__ import annotations
+
+from typing import Literal, Union, get_args
+
+InputBlockType = Literal[
+    "input-checkbox",
+    "input-date",
+    "input-date-range",
+    "input-file",
+    "input-select",
+    "input-slider",
+    "input-text",
+    "input-textarea",
+]
+RunStatus = Literal[
+    "pending", "running", "success", "error", "internal_error", "stopped"
+]
+SnapshotStatus = Literal["pending", "available", "unavailable"]
+StorageMode = Literal["read_write", "readonly"]
+InputValue = Union[str, bool, list[str]]
+
+INPUT_BLOCK_TYPES: frozenset[InputBlockType] = frozenset(get_args(InputBlockType))
+TERMINAL_RUN_STATUSES: frozenset[RunStatus] = frozenset(
+    {"success", "error", "internal_error", "stopped"}
+)