pytupli.storage.TupliAPIClient

class TupliAPIClient[source]

Bases: TupliStorage

Class for storing StorableObjects in the API.

This class provides methods for interacting with the Tupli API, including user management, benchmark operations, artifact handling, and episode management.

User Management

signup(username: str, password: str) -> User: Creates a new user account.
login(username: str, password: str, url: str | None = None) -> None: Authenticates with the API and stores access tokens.
list_users() -> list[User]: Lists all users.
list_roles() -> list[UserRole]: Lists all available user roles.
change_password(username: str, new_password: str) -> None: Changes a user’s password.
change_roles(username: str, roles: list[str]) -> None: Changes a user’s roles.
delete_user(username: str) -> None: Deletes a user and their content.

Benchmark Operations

store_benchmark(benchmark_query: BenchmarkQuery) -> BenchmarkHeader: Saves a benchmark to the API.
load_benchmark(uri: str) -> Benchmark: Loads a benchmark from the API.
list_benchmarks(filter: BaseFilter = None) -> list[BenchmarkHeader]: Lists benchmarks matching the filter.
delete_benchmark(uri: str) -> None: Deletes a benchmark.
publish_benchmark(uri: str) -> None: Publishes a benchmark.

Artifact Operations

store_artifact(artifact: bytes, metadata: ArtifactMetadata) -> ArtifactMetadataItem: Stores an artifact in the API.
load_artifact(uri: str, **kwargs) -> bytes: Loads an artifact from the API.
list_artifacts(filter: BaseFilter = None) -> list[ArtifactMetadataItem]: Lists artifacts matching the filter.
delete_artifact(uri: str) -> None: Deletes an artifact.
publish_artifact(uri: str) -> None: Publishes an artifact.

Episode Operations

record_episode(episode: Episode) -> EpisodeHeader: Records an episode in the API.
publish_episode(uri: str) -> None: Publishes an episode.
list_episodes(filter: BaseFilter = None, include_tuples: bool = False) -> list[EpisodeHeader] | list[EpisodeItem]: Lists episodes matching the filter.
delete_episode(uri: str) -> None: Deletes an episode.

Configuration()

set_url(url: str) -> None: Sets the base URL for the API.

Methods

`change_password`	Changes a user's password.
`change_roles`	Changes a user's roles.
`delete_artifact`	Deletes the specified artifact from the API.
`delete_benchmark`	Deletes the specified object from the API.
`delete_episode`	Deletes an episode from the API.
`delete_user`	Deletes a user and all their content.
`list_artifacts`	Lists all artifacts in the storage that match the specified filter.
`list_benchmarks`	Lists all benchmarks in the storage that match the specified filter.
`list_episodes`	Lists all episodes in the API that match the specified filter.
`list_roles`	Lists all user roles.
`list_users`	Lists all users.
`load_artifact`	Load artifact from the API.
`load_benchmark`	Loads the serialized benchmark from the API.
`login`	Authenticates with the API and stores the access and refresh tokens.
`publish_artifact`	Publishes the artifact in the API.
`publish_benchmark`	Publishes the benchmark in the API.
`publish_episode`	Publishes an episode in the API.
`record_episode`	Records an episode in the API.
`set_url`	Sets the base URL for the API.
`signup`	Creates a new user account.
`store_artifact`	Stores the artifact in the API.
`store_benchmark`	Saves the serialized object to the API.

_authenticated_request(method, url, **kwargs) → Response[source]

Executes an authenticated request to the API. Handles token refresh if the access token is expired.

Parameters:

method (str) – HTTP method (get, post, put, delete)
url (str) – URL for the request
**kwargs – Additional arguments for the request

Returns:

Response – The response from the request

Raises:

TupliStorageError – If the request fails or the token refresh fails.

_get_bearer_token() → dict[source]

Gets the bearer token for API requests. First tries to use the stored access token, and if that fails, tries to refresh the token.

Returns:: dict – Headers with the bearer token.

_refresh_token() → dict[source]

Refreshes the access token using the stored refresh token.

Returns:: dict – Headers with the refreshed bearer token.
Raises:: TupliStorageError – If the refresh token is not available or the refresh fails.

change_password(username: str, new_password: str) → None[source]

Changes a user’s password.

Parameters:

username (str) – The username of the account to change
new_password (str) – The new password

change_roles(username: str, roles: list[str]) → None[source]

Changes a user’s roles.

Parameters:

username (str) – The username of the account to change
roles (list[str]) – The list of new roles

delete_artifact(uri: str) → None[source]

Deletes the specified artifact from the API.

Parameters:: uri (str) – The hash of the artifact to be deleted.

delete_benchmark(uri: str) → None[source]

Deletes the specified object from the API.

Parameters:: uri (str) – The hash of the object to be deleted.

delete_episode(uri: str) → None[source]

Deletes an episode from the API.

Parameters:: uri (str) – The ID of the episode to delete.

delete_user(username: str) → None[source]

Deletes a user and all their content.

Parameters:: username (str) – The username of the account to delete

list_artifacts(filter: BaseFilter = None) → list[ArtifactMetadataItem][source]

Lists all artifacts in the storage that match the specified filter.

Parameters:: filter (BaseFilter, optional) – The filter to apply when listing artifacts.
Returns:: list[ArtifactMetadataItem] – A list of artifacts that match the filter.

list_benchmarks(filter: BaseFilter = None) → list[BenchmarkHeader][source]

Lists all benchmarks in the storage that match the specified filter.

Parameters:: filter (BaseFilter, optional) – The filter to apply when listing benchmarks.
Returns:: list[BenchmarkHeader] – A list of benchmark headers that match the filter.

list_episodes(filter: BaseFilter = None, include_tuples: bool = False) → list[EpisodeHeader] | list[EpisodeItem][source]

Lists all episodes in the API that match the specified filter.

Parameters:

filter (BaseFilter, optional) – The filter to apply when listing episodes.
include_tuples (bool, optional) – Whether to include tuples in the episode data.

Returns:

list[EpisodeHeader] | list[EpisodeItem] – A list of episode headers or full episode items.

list_roles() → list[UserRole][source]

Lists all user roles.

Returns:: list[UserRole] – A list of all user roles

list_users() → list[User][source]

Lists all users.

Returns:: list[User] – A list of all users

load_artifact(uri: str, **kwargs) → bytes[source]

Load artifact from the API.

Parameters:: uri (str) – hash of the object to be loaded.
Returns:: bytes – The raw artifact data

load_benchmark(uri: str) → Benchmark[source]

Loads the serialized benchmark from the API.

Parameters:: uri (str) – hash of the object to be loaded.
Returns:: Benchmark – The loaded benchmark object.

login(username: str, password: str, url: str | None = None) → None[source]

Authenticates with the API and stores the access and refresh tokens.

Parameters:

username (str) – The username for the API.
password (str) – The password for the API.
url (str, optional) – The base URL for the API. If not provided, uses the stored URL. If provided, it will be set as the new base URL.

publish_artifact(uri: str) → None[source]

Publishes the artifact in the API.

Parameters:: uri (str) – The hash of the artifact to be published.

publish_benchmark(uri: str) → None[source]

Publishes the benchmark in the API.

Parameters:: uri (str) – The hash of the benchmark to be published.

publish_episode(uri: str) → None[source]

Publishes an episode in the API.

Parameters:: uri (str) – The ID of the episode to publish.

record_episode(episode: Episode) → EpisodeHeader[source]

Records an episode in the API.

Parameters:: episode (Episode) – The episode to record.
Returns:: EpisodeHeader – The header of the recorded episode.

set_url(url: str) → None[source]

Sets the base URL for the API.

Parameters:: url (str) – The base URL for the API.

signup(username: str, password: str) → User[source]

Creates a new user account.

Parameters:

username (str) – The username for the new account
password (str) – The password for the new account

Returns:

User – The created user object

store_artifact(artifact: bytes, metadata: ArtifactMetadata) → ArtifactMetadataItem[source]

Stores the artifact in the API.

Parameters:

artifact (bytes) – The artifact to store.
metadata (dict, optional) – Metadata for the artifact.

Returns:

ArtifactMetadataItem – Metadata item for the stored artifact.

store_benchmark(benchmark_query: BenchmarkQuery) → BenchmarkHeader[source]

Saves the serialized object to the API.

Parameters:: benchmark_query (BenchmarkQuery) – The serialized benchmark object to be saved as well as some metadata.
Returns:: BenchmarkHeader – The header of the saved benchmark.