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, group management, role 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.
change_password(username: str, new_password: str) -> None: Changes a user’s password.
delete_user(username: str) -> None: Deletes a user and their content.

Group Management

create_group(group: Group) -> Group: Creates a new group.
list_groups() -> list[Group]: Lists all groups accessible to the current user.
read_group(group_name: str) -> GroupWithMembers: Reads a specific group with its members.
delete_group(group_name: str) -> None: Deletes a group.
add_group_members(group_membership_query: GroupMembershipQuery) -> GroupWithMembers: Adds members to a group with specified roles.
remove_group_members(group_membership_query: GroupMembershipQuery) -> GroupWithMembers: Removes members from a group.

Role Management

list_roles() -> list[UserRole]: Lists all available user roles.
create_role(role: UserRole) -> UserRole: Creates a new role.
delete_role(role_name: str) -> None: Deletes a role.

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

`add_group_members`	Adds members to a group with specified roles.
`change_password`	Changes a user's password.
`create_group`	Creates a new group.
`create_role`	Creates a new role.
`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_group`	Deletes a group.
`delete_role`	Deletes a role.
`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_groups`	Lists all groups accessible to the current user.
`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.
`read_group`	Reads a specific group with its members.
`record_episode`	Records an episode in the API.
`remove_group_members`	Removes members from a group.
`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.
`unpublish_artifact`	Unpublishes an artifact from the specified group.
`unpublish_benchmark`	Unpublishes a benchmark from the specified group.
`unpublish_episode`	Unpublishes an episode from the specified group.

_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.

add_group_members(group_membership_query: GroupMembershipQuery) → GroupWithMembers[source]

Adds members to a group with specified roles.

Parameters:: group_membership_query (GroupMembershipQuery) – The membership query specifying group and members
Returns:: GroupWithMembers – The updated group with members

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

create_group(group: Group) → Group[source]

Creates a new group.

Parameters:: group (Group) – The group to create
Returns:: Group – The created group object

create_role(role: UserRole) → UserRole[source]

Creates a new role.

Parameters:: role (UserRole) – The role to create
Returns:: UserRole – The created role object

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_group(group_name: str) → None[source]

Deletes a group.

Parameters:: group_name (str) – The name of the group to delete

delete_role(role_name: str) → None[source]

Deletes a role.

Parameters:: role_name (str) – The name of the role 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_groups() → list[Group][source]

Lists all groups accessible to the current user.

Returns:: list[Group] – A list of groups

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, publish_in: str = 'global') → None[source]

Publishes the artifact in the API.

Parameters:

uri (str) – The hash of the artifact to be published.
publish_in (str) – The group to publish the artifact in. Defaults to ‘global’.

publish_benchmark(uri: str, publish_in: str = 'global') → None[source]

Publishes the benchmark in the API.

Parameters:

uri (str) – The hash of the benchmark to be published.
publish_in (str) – The group to publish the benchmark in. Defaults to ‘global’.

publish_episode(uri: str, publish_in: str = 'global') → None[source]

Publishes an episode in the API.

Parameters:

uri (str) – The ID of the episode to publish.
publish_in (str) – The group to publish the episode in. Defaults to ‘global’.

read_group(group_name: str) → GroupWithMembers[source]

Reads a specific group with its members.

Parameters:: group_name (str) – The name of the group to read
Returns:: GroupWithMembers – The group with its members

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.

remove_group_members(group_membership_query: GroupMembershipQuery) → GroupWithMembers[source]

Removes members from a group.

Parameters:: group_membership_query (GroupMembershipQuery) – The membership query specifying group and members
Returns:: GroupWithMembers – The updated group with members

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.

unpublish_artifact(uri: str, unpublish_from: str) → None[source]

Unpublishes an artifact from the specified group.

Parameters:

uri (str) – The ID of the artifact to unpublish.
unpublish_from (str) – The group to unpublish the artifact from.

unpublish_benchmark(uri: str, unpublish_from: str) → None[source]

Unpublishes a benchmark from the specified group.

Parameters:

uri (str) – The ID of the benchmark to unpublish.
unpublish_from (str) – The group to unpublish the benchmark from.

unpublish_episode(uri: str, unpublish_from: str) → None[source]

Unpublishes an episode from the specified group.

Parameters:

uri (str) – The ID of the episode to unpublish.
unpublish_from (str) – The group to unpublish the episode from.