Skip to content

RestClient

class tinker.RestClient(holder)

Client for REST API operations like listing checkpoints and metadata.

The RestClient provides access to various REST endpoints for querying model information, checkpoints, and other resources. You typically get one by calling service_client.create_rest_client().

Key methods:

  • list_checkpoints() - list available model checkpoints (both training and sampler)
  • list_user_checkpoints() - list all checkpoints across all user's training runs
  • get_training_run() - get model information and metadata as ModelEntry
  • delete_checkpoint() - delete an existing checkpoint for a training run
  • get_checkpoint_archive_url() - get signed URL to download checkpoint archive
  • publish_checkpoint_from_tinker_path() - publish a checkpoint to make it public
  • unpublish_checkpoint_from_tinker_path() - unpublish a checkpoint to make it private
  • set_checkpoint_ttl_from_tinker_path() - set or remove TTL on a checkpoint
  • assign_session_project() - move a session into a project
rest_client = service_client.create_rest_client()
training_run = rest_client.get_training_run("run-id").result()
print(f"Training Run: {training_run.training_run_id}, LoRA: {training_run.is_lora}")
checkpoints = rest_client.list_checkpoints("run-id").result()
print(f"Found {len(checkpoints.checkpoints)} checkpoints")
for checkpoint in checkpoints.checkpoints:
print(f"  {checkpoint.checkpoint_type}: {checkpoint.checkpoint_id}")

Parameters:

get_training_run(training_run_id, access_scope='owned')

Get training run info.

Parameters:

Returns: A Future containing the training run information

future = rest_client.get_training_run("run-id")
response = future.result()
print(f"Training Run ID: {response.training_run_id}, Base: {response.base_model}")

Async variant: get_training_run_async()

get_training_run_by_tinker_path(tinker_path, access_scope='owned')

Get training run info.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint
  • access_scope (Literal['owned', 'accessible'], default: 'owned')

Returns: A Future containing the training run information

future = rest_client.get_training_run_by_tinker_path("tinker://run-id/weights/checkpoint-001")
response = future.result()
print(f"Training Run ID: {response.training_run_id}, Base: {response.base_model}")

Async variant: get_training_run_by_tinker_path_async()

get_weights_info_by_tinker_path(tinker_path)

Get checkpoint information from a tinker path.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint

Returns: An APIFuture containing the checkpoint information. The future is awaitable.

future = rest_client.get_weights_info_by_tinker_path("tinker://run-id/weights/checkpoint-001")
response = future.result()  # or await future
print(f"Base Model: {response.base_model}, LoRA Rank: {response.lora_rank}")

list_training_runs(limit=20, offset=0, access_scope='owned')

List training runs with pagination support.

Parameters:

  • limit (int, default: 20) – Maximum number of training runs to return (default 20)
  • offset (int, default: 0) – Offset for pagination (default 0)
  • access_scope (Literal['owned', 'accessible'], default: 'owned')

Returns: A Future containing the TrainingRunsResponse with training runs and cursor info

future = rest_client.list_training_runs(limit=50)
response = future.result()
print(f"Found {len(response.training_runs)} training runs")
print(f"Total: {response.cursor.total_count}")
# Get next page
next_page = rest_client.list_training_runs(limit=50, offset=50)

Async variant: list_training_runs_async()

list_checkpoints(training_run_id)

List available checkpoints (both training and sampler).

Parameters:

Returns: A Future containing the CheckpointsListResponse with available checkpoints

future = rest_client.list_checkpoints("run-id")
response = future.result()
for checkpoint in response.checkpoints:
if checkpoint.checkpoint_type == "training":
print(f"Training checkpoint: {checkpoint.checkpoint_id}")
elif checkpoint.checkpoint_type == "sampler":
print(f"Sampler checkpoint: {checkpoint.checkpoint_id}")

Async variant: list_checkpoints_async()

get_checkpoint_archive_url(training_run_id, checkpoint_id)

Get signed URL to download checkpoint archive.

Parameters:

Returns: A Future containing the CheckpointArchiveUrlResponse with signed URL and expiration

future = rest_client.get_checkpoint_archive_url("run-id", "checkpoint-123")
response = future.result()
print(f"Download URL: {response.url}")
print(f"Expires at: {response.expires_at}")
# Use the URL to download the archive with your preferred HTTP client

Async variant: get_checkpoint_archive_url_async()

delete_checkpoint(training_run_id, checkpoint_id)

Delete a checkpoint for a training run.

Parameters:

Returns: ConcurrentFuture[None]

Async variant: delete_checkpoint_async()

delete_checkpoint_from_tinker_path(tinker_path)

Delete a checkpoint referenced by a tinker path.

Parameters:

Returns: ConcurrentFuture[None]

Async variant: delete_checkpoint_from_tinker_path_async()

get_audit_log(event_type='all', day=None)

Get an audit log of events for the caller's organization.

Requires the tinker-admin RBAC role (VIEW_AUDIT_LOG capability).

Parameters:

  • event_type (Literal['all', 'checkpoints'], default: 'all') – Type of events to include. "all" and "checkpoints" are currently equivalent. Defaults to "all".
  • day (date | None, default: None) – The date to query (default: today). The window covers midnight to midnight UTC.

Returns: A Future containing the AuditLogResponse with audit log entries

from datetime import date

future = rest_client.get_audit_log()
response = future.result()
print(f"Found {len(response.entries)} audit entries")
for entry in response.entries:
print(f"  {entry.timestamp}: {entry.event} ({entry.tinker_path})")

# Query a specific day
future = rest_client.get_audit_log(day=date(2025, 1, 15))

Async variant: get_audit_log_async()

get_checkpoint_archive_url_from_tinker_path(tinker_path)

Get signed URL to download checkpoint archive.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint

Returns: A Future containing the CheckpointArchiveUrlResponse with signed URL and expiration

Async variant: get_checkpoint_archive_url_from_tinker_path_async()

publish_checkpoint_from_tinker_path(tinker_path)

Publish a checkpoint referenced by a tinker path to make it publicly accessible.

Only the exact owner of the training run can publish checkpoints. Published checkpoints can be unpublished using the unpublish_checkpoint_from_tinker_path method.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint (e.g., "tinker://run-id/weights/0001")

Returns: A Future that completes when the checkpoint is published

future = rest_client.publish_checkpoint_from_tinker_path("tinker://run-id/weights/0001")
future.result()  # Wait for completion
print("Checkpoint published successfully")

Async variant: publish_checkpoint_from_tinker_path_async()

unpublish_checkpoint_from_tinker_path(tinker_path)

Unpublish a checkpoint referenced by a tinker path to make it private again.

Only the exact owner of the training run can unpublish checkpoints. This reverses the effect of publishing a checkpoint.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint (e.g., "tinker://run-id/weights/0001")

Returns: A Future that completes when the checkpoint is unpublished

future = rest_client.unpublish_checkpoint_from_tinker_path("tinker://run-id/weights/0001")
future.result()  # Wait for completion
print("Checkpoint unpublished successfully")

Async variant: unpublish_checkpoint_from_tinker_path_async()

set_checkpoint_ttl_from_tinker_path(tinker_path, ttl_seconds)

Set or remove the TTL on a checkpoint referenced by a tinker path.

If ttl_seconds is provided, the checkpoint will expire after that many seconds from now. If ttl_seconds is None, any existing expiration will be removed.

Parameters:

  • tinker_path (str) – The tinker path to the checkpoint (e.g., "tinker://run-id/weights/0001")
  • ttl_seconds (int | None) – Number of seconds until expiration, or None to remove TTL

Returns: A Future that completes when the TTL is set

future = rest_client.set_checkpoint_ttl_from_tinker_path("tinker://run-id/weights/0001", 86400)
future.result()  # Wait for completion
print("Checkpoint TTL set successfully")

Async variant: set_checkpoint_ttl_from_tinker_path_async()

list_user_checkpoints(limit=100, offset=0)

List all checkpoints for the current user across all their training runs.

This method retrieves checkpoints from all training runs owned by the authenticated user, sorted by time (newest first). It supports pagination for efficiently handling large numbers of checkpoints.

Parameters:

  • limit (int, default: 100) – Maximum number of checkpoints to return (default 100)
  • offset (int, default: 0) – Offset for pagination (default 0)

Returns: A Future containing the CheckpointsListResponse with checkpoints and cursor info

future = rest_client.list_user_checkpoints(limit=50)
response = future.result()
print(f"Found {len(response.checkpoints)} checkpoints")
print(f"Total: {response.cursor.total_count if response.cursor else 'Unknown'}")
for checkpoint in response.checkpoints:
print(f"  {checkpoint.training_run_id}/{checkpoint.checkpoint_id}")
# Get next page if there are more checkpoints
if response.cursor and response.cursor.offset + response.cursor.limit < response.cursor.total_count:
next_page = rest_client.list_user_checkpoints(limit=50, offset=50)

Async variant: list_user_checkpoints_async()

get_session(session_id, access_scope='owned')

Get session information including all training runs and samplers.

Parameters:

  • session_id (str) – The session ID to get information for
  • access_scope (Literal['owned', 'accessible'], default: 'owned')

Returns: A Future containing the GetSessionResponse with training_run_ids and sampler_ids

future = rest_client.get_session("session-id")
response = future.result()
print(f"Training runs: {len(response.training_run_ids)}")
print(f"Samplers: {len(response.sampler_ids)}")

Async variant: get_session_async()

list_sessions(limit=20, offset=0, access_scope='owned')

List sessions with pagination support.

Parameters:

  • limit (int, default: 20) – Maximum number of sessions to return (default 20)
  • offset (int, default: 0) – Offset for pagination (default 0)
  • access_scope (Literal['owned', 'accessible'], default: 'owned')

Returns: A Future containing the ListSessionsResponse with list of session IDs

future = rest_client.list_sessions(limit=50)
response = future.result()
print(f"Found {len(response.sessions)} sessions")
# Get next page
next_page = rest_client.list_sessions(limit=50, offset=50)

Async variant: list_sessions_async()

assign_session_project(session_id, project_id)

Move a session (and all of its training runs/samplers) into a project.

Use this to attach a previously-created session to a project, or to move a session between projects. Clearing the project is not supported — sessions cannot be moved out of a project once placed.

Parameters:

Returns: A Future that completes when the session has been moved

future = rest_client.assign_session_project("session-id", "project-id")
future.result()

Async variant: assign_session_project_async()

get_sampler(sampler_id)

Get sampler information.

Parameters:

  • sampler_id (str) – The sampler ID (sampling_session_id) to get information for

Returns: An APIFuture containing the GetSamplerResponse with sampler details

# Sync usage
future = rest_client.get_sampler("session-id:sample:0")
response = future.result()
print(f"Base model: {response.base_model}")
print(f"Model path: {response.model_path}")

# Async usage
response = await rest_client.get_sampler("session-id:sample:0")
print(f"Base model: {response.base_model}")

Async variant: get_sampler_async()

Referenced by