Add token lifecycle management to client.tokens (refresh/revoke/scopes/copy)

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- `client.tokens` gains lifecycle management for existing static tokens: `list()`, `get(name)`, `scopes(name)`, `refresh(name)`, `revoke(name)`, and `copy(name)`, matching `tb token ls/rm/refresh/scopes/copy`. `copy()` returns the token's current value since a library has no clipboard to copy it to. Backed by new `TinybirdApi.list_tokens()`/`get_token()`/`refresh_token()`/`revoke_token()` wrapping `/v0/tokens`.
+
 ## [0.4.0] - 2026-06-29
 
 ### Added

README.md

@@ -350,6 +350,40 @@ jwt_token = result["token"]
 - **`fixed_params`**: For pipes, embed parameters that cannot be overridden by the caller.
 - **`filter`**: For datasources, append a SQL WHERE clause (for example, `"org_id = 'acme'"`).
 
+## Token Lifecycle Management
+
+Manage static tokens the workspace already has, matching `tb token ls/rm/refresh/scopes/copy`.
+
+```python
+from tinybird_sdk import create_client
+
+client = create_client(
+    {
+        "base_url": "https://api.tinybird.co",
+        "token": "p.your_admin_token",
+    }
+)
+
+# List tokens (tb token ls)
+tokens = client.tokens.list()
+
+# Get a token's details, including its scopes and current value (tb token get)
+token = client.tokens.get("user_123_session")
+
+# List just a token's scopes (tb token scopes)
+scopes = client.tokens.scopes("user_123_session")
+
+# Rotate a token's value (tb token refresh)
+refreshed = client.tokens.refresh("user_123_session")
+
+# Revoke (delete) a token (tb token rm)
+client.tokens.revoke("user_123_session")
+
+# Get a token's current value (tb token copy copies it to the clipboard;
+# in a library there's no clipboard, so this returns the same value instead)
+value = client.tokens.copy("user_123_session")
+```
+
 ## CLI Commands
 
 This package installs `tinybird` as a runtime dependency.

src/tinybird_sdk/api/api.py

@@ -363,6 +363,72 @@ def create_token(
             self._raise_for_error(response.status_code, response.text)
         return response.json()
 
+    def list_tokens(self, options: dict[str, Any] | None = None) -> dict[str, Any]:
+        """List tokens in the workspace via ``GET /v0/tokens``."""
+        options = options or {}
+        response = self.request(
+            "/v0/tokens",
+            method="GET",
+            token=options.get("token"),
+            timeout=options.get("timeout"),
+        )
+        if not response.ok:
+            self._raise_for_error(response.status_code, response.text)
+        return response.json()
+
+    def get_token(self, token_name: str, options: dict[str, Any] | None = None) -> dict[str, Any]:
+        """Get a single token's details, including its scopes and current value, via
+        ``GET /v0/tokens/{name}``."""
+        options = options or {}
+        response = self.request(
+            f"/v0/tokens/{token_name}",
+            method="GET",
+            token=options.get("token"),
+            timeout=options.get("timeout"),
+        )
+        if not response.ok:
+            self._raise_for_error(response.status_code, response.text)
+        return response.json()
+
+    def refresh_token(
+        self, token_name: str, options: dict[str, Any] | None = None
+    ) -> dict[str, Any]:
+        """Rotate a token's value via ``POST /v0/tokens/{name}/refresh``."""
+        options = options or {}
+        response = self.request(
+            f"/v0/tokens/{token_name}/refresh",
+            method="POST",
+            token=options.get("token"),
+            body="",
+            timeout=options.get("timeout"),
+        )
+        if not response.ok:
+            self._raise_for_error(response.status_code, response.text)
+        return self._json_or_empty(response)
+
+    def revoke_token(
+        self, token_name: str, options: dict[str, Any] | None = None
+    ) -> dict[str, Any]:
+        """Revoke (delete) a token via ``DELETE /v0/tokens/{name}``."""
+        options = options or {}
+        response = self.request(
+            f"/v0/tokens/{token_name}",
+            method="DELETE",
+            token=options.get("token"),
+            timeout=options.get("timeout"),
+        )
+        if not response.ok:
+            self._raise_for_error(response.status_code, response.text)
+        return self._json_or_empty(response)
+
+    def _json_or_empty(self, response: Any) -> dict[str, Any]:
+        if not response.text.strip():
+            return {}
+        try:
+            return response.json()
+        except json.JSONDecodeError:
+            return {}
+
     def _timeout_seconds(self, timeout_ms: int | None) -> float:
         timeout = timeout_ms if timeout_ms is not None else self._default_timeout
         return max(timeout / 1000.0, 0.001)

src/tinybird_sdk/client/tokens.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
-from typing import Any, Callable
+from typing import Any, Callable, cast
+from typing import List as _List
 
+from ..api.api import TinybirdApi, TinybirdApiError
 from ..api.tokens import TokenApiError, create_jwt
-from .types import TinybirdError
+from .types import TinybirdError, TinybirdErrorResponse
 
 
 class TokensNamespace:
@@ -38,3 +40,53 @@ def create_jwt(self, options: dict[str, Any]) -> dict[str, str]:
                     "status": error.status,
                 },
             ) from error
+
+    def list(self, options: dict[str, Any] | None = None) -> _List[dict[str, Any]]:
+        """List tokens in the workspace, matching ``tb token ls``."""
+        result = self._request(lambda api, opts: api.list_tokens(opts), options)
+        return list(result.get("tokens", []))
+
+    def get(self, name: str, options: dict[str, Any] | None = None) -> dict[str, Any]:
+        """Get a single token's details, including its scopes and current value."""
+        return self._request(lambda api, opts: api.get_token(name, opts), options)
+
+    def scopes(self, name: str, options: dict[str, Any] | None = None) -> _List[dict[str, Any]]:
+        """List a token's scopes, matching ``tb token scopes``."""
+        return list(self.get(name, options).get("scopes", []))
+
+    def refresh(self, name: str, options: dict[str, Any] | None = None) -> dict[str, Any]:
+        """Rotate a token's value, matching ``tb token refresh``."""
+        return self._request(lambda api, opts: api.refresh_token(name, opts), options)
+
+    def revoke(self, name: str, options: dict[str, Any] | None = None) -> dict[str, Any]:
+        """Revoke (delete) a token, matching ``tb token rm``."""
+        return self._request(lambda api, opts: api.revoke_token(name, opts), options)
+
+    def copy(self, name: str, options: dict[str, Any] | None = None) -> str:
+        """Return a token's current value.
+
+        ``tb token copy`` copies the token value to the system clipboard, which has
+        no equivalent in a library context; this returns the same underlying value
+        (via ``get``) for the caller to use or store as needed.
+        """
+        return str(self.get(name, options).get("token", ""))
+
+    def _request(
+        self,
+        call: "Callable[[TinybirdApi, dict[str, Any]], dict[str, Any]]",
+        options: dict[str, Any] | None,
+    ) -> dict[str, Any]:
+        opts = dict(options or {})
+        opts.setdefault("timeout", self._timeout)
+        api = TinybirdApi(
+            {
+                "base_url": self._base_url,
+                "token": self._get_token(),
+                "timeout": self._timeout,
+            }
+        )
+        try:
+            return call(api, opts)
+        except TinybirdApiError as error:
+            response = cast(TinybirdErrorResponse | None, error.response)
+            raise TinybirdError(str(error), error.status_code, response) from error

tests/test_token_lifecycle.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import pytest
+
+import tinybird_sdk.api.api as api_module
+from tinybird_sdk.client.base import TinybirdClient
+from tinybird_sdk.client.types import TinybirdError
+
+
+class _FakeResponse:
+    def __init__(self, status_code: int, payload: Any = None, text: str | None = None):
+        self.status_code = status_code
+        self._payload = payload
+        self.text = (
+            text if text is not None else (json.dumps(payload) if payload is not None else "")
+        )
+
+    @property
+    def ok(self) -> bool:
+        return 200 <= self.status_code < 300
+
+    def json(self) -> Any:
+        return self._payload
+
+
+def _make_client() -> TinybirdClient:
+    return TinybirdClient({"base_url": "https://api.tinybird.co", "token": "p.workspace"})
+
+
+def _capture_fetch(captured: dict[str, Any], response: _FakeResponse):
+    def fake_fetch(url: str, **kwargs: Any) -> _FakeResponse:
+        captured["url"] = url
+        captured["method"] = kwargs.get("method")
+        captured["headers"] = kwargs.get("headers")
+        captured["body"] = kwargs.get("body")
+        return response
+
+    return fake_fetch
+
+
+def test_tokens_list_unwraps_tokens_key(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, Any] = {}
+    monkeypatch.setattr(
+        api_module,
+        "tinybird_fetch",
+        _capture_fetch(captured, _FakeResponse(200, {"tokens": [{"name": "t1"}, {"name": "t2"}]})),
+    )
+
+    result = _make_client().tokens.list()
+
+    assert result == [{"name": "t1"}, {"name": "t2"}]
+    assert captured["method"] == "GET"
+    assert captured["url"].endswith("/v0/tokens")
+
+
+def test_tokens_get_returns_full_token(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, Any] = {}
+    payload = {"name": "t1", "token": "p.abc", "scopes": [{"type": "DATASOURCES:READ"}]}
+    monkeypatch.setattr(
+        api_module, "tinybird_fetch", _capture_fetch(captured, _FakeResponse(200, payload))
+    )
+
+    result = _make_client().tokens.get("t1")
+
+    assert result == payload
+    assert captured["url"].endswith("/v0/tokens/t1")
+
+
+def test_tokens_scopes_extracts_scopes_field(monkeypatch: pytest.MonkeyPatch) -> None:
+    payload = {"name": "t1", "token": "p.abc", "scopes": [{"type": "DATASOURCES:READ"}]}
+    monkeypatch.setattr(
+        api_module, "tinybird_fetch", _capture_fetch({}, _FakeResponse(200, payload))
+    )
+
+    assert _make_client().tokens.scopes("t1") == [{"type": "DATASOURCES:READ"}]
+
+
+def test_tokens_refresh_posts_to_refresh_endpoint(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, Any] = {}
+    monkeypatch.setattr(
+        api_module,
+        "tinybird_fetch",
+        _capture_fetch(captured, _FakeResponse(200, {"name": "t1", "token": "p.new"})),
+    )
+
+    result = _make_client().tokens.refresh("t1")
+
+    assert result == {"name": "t1", "token": "p.new"}
+    assert captured["method"] == "POST"
+    assert captured["url"].endswith("/v0/tokens/t1/refresh")
+
+
+def test_tokens_revoke_deletes_and_tolerates_empty_body(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: dict[str, Any] = {}
+    monkeypatch.setattr(
+        api_module, "tinybird_fetch", _capture_fetch(captured, _FakeResponse(200, text=""))
+    )
+
+    result = _make_client().tokens.revoke("t1")
+
+    assert result == {}
+    assert captured["method"] == "DELETE"
+    assert captured["url"].endswith("/v0/tokens/t1")
+
+
+def test_tokens_copy_returns_current_value(monkeypatch: pytest.MonkeyPatch) -> None:
+    payload = {"name": "t1", "token": "p.secret-value"}
+    monkeypatch.setattr(
+        api_module, "tinybird_fetch", _capture_fetch({}, _FakeResponse(200, payload))
+    )
+
+    assert _make_client().tokens.copy("t1") == "p.secret-value"
+
+
+def test_tokens_methods_wrap_api_errors(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        api_module,
+        "tinybird_fetch",
+        _capture_fetch({}, _FakeResponse(403, {"error": "Forbidden"})),
+    )
+
+    client = _make_client()
+    with pytest.raises(TinybirdError, match="Forbidden"):
+        client.tokens.get("t1")
+    with pytest.raises(TinybirdError, match="Forbidden"):
+        client.tokens.refresh("t1")
+    with pytest.raises(TinybirdError, match="Forbidden"):
+        client.tokens.revoke("t1")