Commands

The commands layer implements the business logic for the Kelvin CLI. Each command class orchestrates services to perform operations, defines result data models, and raises domain-specific exceptions.

Commands are instantiated via the dependency injection container (kelvin.sdk.container.Services) and receive their dependencies through constructor injection.

Shared Utilities

Common constants and helpers used across command modules.

Shared constants and helpers for command modules.

kelvin.sdk.commands._shared.format_datetime(dt)[source]

Format datetime for display.

Parameters:

dt (Optional[datetime]) – The datetime to format.

Return type:

Optional[str]

Returns:

Formatted string or None.

App Commands

Business logic for local application creation, building, and uploading.

App command business logic.

class kelvin.sdk.commands.app.AppCreateResult(app_name, app_path, template_name, files_created, symlinks_created)[source]

Bases: object

Result of app creation.

Parameters:
app_name: str
app_path: Path
template_name: str
files_created: list[str]
class kelvin.sdk.commands.app.TemplateListResult(templates)[source]

Bases: object

Result of template list operation.

Parameters:

templates (list[TemplateInfo])

templates: list[TemplateInfo]
class kelvin.sdk.commands.app.LegacyDockerfileGenerated(dockerfile_path, entry_point, python_version, builder_version)[source]

Bases: object

Information about a generated legacy Dockerfile.

Parameters:

  • dockerfile_path (Path)

  • entry_point (str)

  • python_version (str)

  • builder_version (str)

dockerfile_path: Path
entry_point: str
python_version: str
builder_version: str
class kelvin.sdk.commands.app.AppBuildResult(app_name, app_version, image_name, platforms, legacy_dockerfile_generated=None)[source]

Bases: object

Result of app build operation.

Parameters:
app_name: str
app_version: str
image_name: str
platforms: list[str]
legacy_dockerfile_generated: LegacyDockerfileGenerated | None = None
class kelvin.sdk.commands.app.AppUploadResult(app_name, app_version, image_name, platforms, legacy_dockerfile_generated=None)[source]

Bases: object

Result of app upload operation.

Parameters:
app_name: str
app_version: str
image_name: str
platforms: list[str]
legacy_dockerfile_generated: LegacyDockerfileGenerated | None = None
exception kelvin.sdk.commands.app.AppCommandsError(message=None, exit_code=None)[source]

Bases: CLIError

Base exception for app command operations.

Parameters:

  • message (str)

  • exit_code (int)

exception kelvin.sdk.commands.app.AppAlreadyExistsError(app_path)[source]

Bases: AppCommandsError

Raised when trying to create an app that already exists.

Parameters:

app_path (Path)

exit_code: int = 73
app_path: Path
exception kelvin.sdk.commands.app.InvalidAppNameError(app_name, reason)[source]

Bases: AppCommandsError

Raised when app name is invalid.

Parameters:
exit_code: int = 64
app_name: str
exception kelvin.sdk.commands.app.AppConfigNotFoundError(app_dir)[source]

Bases: AppCommandsError

Raised when app config file is not found.

Parameters:

app_dir (Path)

exit_code: int = 66
app_dir: Path
exception kelvin.sdk.commands.app.DockerfileNotFoundError(app_dir)[source]

Bases: AppCommandsError

Raised when Dockerfile is not found.

Parameters:

app_dir (Path)

exit_code: int = 66
app_dir: Path
exception kelvin.sdk.commands.app.AppVersionExistsError(app_name, app_version)[source]

Bases: AppCommandsError

Raised when app version already exists on the platform.

Parameters:

  • app_name (str)

  • app_version (str)

exit_code: int = 73
app_name: str
app_version: str
exception kelvin.sdk.commands.app.RegistryNotConfiguredError[source]

Bases: AppCommandsError

Raised when Docker registry is not configured.

Return type:

None

exit_code: int = 78
exception kelvin.sdk.commands.app.AppManifestError(message)[source]

Bases: AppCommandsError

Raised when app manifest generation fails.

Parameters:

message (str)

exit_code: int = 65
exception kelvin.sdk.commands.app.SchemaValidationFailedError(errors)[source]

Bases: AppCommandsError

Raised when app config fails schema validation.

Parameters:

errors (list[str])

Return type:

None

exit_code: int = 65
validation_errors: list[str]
class kelvin.sdk.commands.app.AppCommands(template_service, docker_service, schema_service)[source]

Bases: object

App command business logic.

Handles application creation, build, upload, and management operations. Receives complete, validated inputs from CLI layer.

Note: api_client and registry_url are passed to upload() method directly to avoid requiring authentication for local-only operations (create, build).

Parameters:
list_templates()[source]

List available application templates.

Return type:

TemplateListResult

create_app(app_name, *, template='app', app_dir=None, description=None, version='0.1.0')[source]

Create a new application from a template.

Parameters:

  • app_name (str) – Name of the application (used as directory name).

  • template (str) – Template to use (default: “app”).

  • app_dir (Optional[Path]) – Parent directory for the app (default: current directory).

  • description (Optional[str]) – Application description (default: app_name).

  • version (str) – Application version (default: “0.1.0”).

Return type:

AppCreateResult

Returns:

AppCreateResult with paths to created files.

Raises:
preview_app(app_name, *, template='app')[source]

Preview files that would be created.

Parameters:

  • app_name (str) – Name of the application.

  • template (str) – Template to use.

Return type:

list[str]

Returns:

List of relative file paths that would be created.

build(app_dir=None, *, build_args=None, architectures=None, fresh=False, skip_schema_validation=False)[source]

Build an application into a Docker image.

Reads the app configuration from app.yaml, validates that a Dockerfile exists, and builds the Docker image using buildx.

For legacy apps without a Dockerfile, one is automatically generated from the legacy template using the entry_point from app.yaml.

Parameters:

  • app_dir (Optional[Path]) – Directory containing the application (default: current directory).

  • build_args (Optional[dict[str, str]]) – Docker build arguments.

  • architectures (Optional[list[str]]) – Target architectures (default: [“amd64”]).

  • fresh (bool) – If True, build without cache.

  • skip_schema_validation (bool) – If True, skip JSON schema validation.

Return type:

AppBuildResult

Returns:

AppBuildResult with build information.

Raises:
upload(api_client, registry_url, app_dir=None, *, build_args=None, architectures=None, fresh=False, skip_schema_validation=False, config_injection_callback=None)[source]

Build and upload an application to the platform.

This performs the following steps: 1. Loads app configuration from app.yaml 2. Checks if the app version already exists on the platform 3. Builds the Docker image and pushes to registry using buildx –push 4. Creates the app version on the platform via API

Parameters:

  • api_client (Client) – Authenticated API client.

  • registry_url (str) – Docker registry URL.

  • app_dir (Optional[Path]) – Directory containing the application (default: current directory).

  • build_args (Optional[dict[str, str]]) – Docker build arguments.

  • architectures (Optional[list[str]]) – Target architectures (default: [“amd64”]).

  • fresh (bool) – If True, build without cache.

  • skip_schema_validation (bool) – If True, skip JSON schema validation.

  • config_injection_callback (Optional[Callable[[dict[str, object]], bool]]) – Optional callback to decide whether to inject config.yaml into manifest defaults. Receives the parsed config dict, returns True to inject.

Return type:

AppUploadResult

Returns:

AppUploadResult with upload information.

Raises:

App Image Commands

Business logic for local Docker image management operations.

App images command business logic.

Handles local application image management operations. These are Docker-only operations that don’t require authentication.

class kelvin.sdk.commands.app_images.LocalImage(name, version, tag, created_at, size)[source]

Bases: object

Local application image information.

Parameters:
name: str
version: str
tag: str
created_at: str | None
size: int
class kelvin.sdk.commands.app_images.ImageListResult(images)[source]

Bases: object

Result of listing local images.

Parameters:

images (list[LocalImage])

images: list[LocalImage]
class kelvin.sdk.commands.app_images.ImageRemoveResult(image_name, removed)[source]

Bases: object

Result of removing a local image.

Parameters:
image_name: str
removed: bool
class kelvin.sdk.commands.app_images.ImageUnpackResult(image_name, container_path, output_path)[source]

Bases: object

Result of unpacking an image.

Parameters:

  • image_name (str)

  • container_path (str)

  • output_path (str)

image_name: str
container_path: str
output_path: str
exception kelvin.sdk.commands.app_images.AppImageError(message=None, exit_code=None)[source]

Bases: CLIError

Base exception for app image operations.

Parameters:

  • message (str)

  • exit_code (int)

exception kelvin.sdk.commands.app_images.InvalidImageNameError(image_name)[source]

Bases: AppImageError

Raised when image name format is invalid.

Parameters:

image_name (str)

exit_code: int = 64
image_name: str
class kelvin.sdk.commands.app_images.AppImageCommands(docker_service)[source]

Bases: object

App image command business logic.

Handles local Docker image operations. These are local-only operations that don’t require authentication to the platform.

Parameters:

docker_service (DockerService)

list_images()[source]

List all locally built application images.

Lists Docker images that were built by KSDK (identified by labels).

Return type:

ImageListResult

Returns:

ImageListResult containing list of local images.

remove_image(app_name_with_version)[source]

Remove an application image from the local registry.

Parameters:

app_name_with_version (str) – Image name with version (e.g., “my-app:1.0.0”).

Return type:

ImageRemoveResult

Returns:

ImageRemoveResult indicating success.

Raises:
unpack_image(app_name_with_version, output_dir, container_dir=None)[source]

Extract content from an application image.

Extracts files from a Docker image to a local directory.

Parameters:

  • app_name_with_version (str) – Image name with version (e.g., “my-app:1.0.0”).

  • output_dir (Path) – Directory to extract files to.

  • container_dir (Optional[str]) – Path inside container to extract (default: /app).

Return type:

ImageUnpackResult

Returns:

ImageUnpackResult with extraction details.

Raises:

Apps Commands

Business logic for platform application registry operations.

Apps command business logic.

Platform application registry operations. These commands interact with the API to manage apps on the platform.

class kelvin.sdk.commands.apps.AppInfo(name, title, app_type, latest_version, updated_at)[source]

Bases: object

Application information.

Parameters:

  • name (str)

  • title (str | None)

  • app_type (str | None)

  • latest_version (str | None)

  • updated_at (str | None)

name: str
title: str | None
app_type: str | None
latest_version: str | None
updated_at: str | None
class kelvin.sdk.commands.apps.AppListResult(apps, total)[source]

Bases: object

Result of listing apps.

Parameters:
apps: list[AppInfo]
total: int
class kelvin.sdk.commands.apps.AppVersionInfo(version, updated_at)[source]

Bases: object

Application version information.

Parameters:

  • version (str)

  • updated_at (str | None)

version: str
updated_at: str | None
class kelvin.sdk.commands.apps.AppDetailResult(name, title, app_type, description, latest_version, created_at, updated_at, versions)[source]

Bases: object

Result of showing app details.

Parameters:
name: str
title: str | None
app_type: str | None
description: str | None
latest_version: str | None
created_at: str | None
updated_at: str | None
versions: list[AppVersionInfo]
class kelvin.sdk.commands.apps.AppVersionDetailResult(name, version, title, app_type, description, created_at, updated_at)[source]

Bases: object

Result of showing app version details.

Parameters:

  • name (str)

  • version (str)

  • title (str | None)

  • app_type (str | None)

  • description (str | None)

  • created_at (str | None)

  • updated_at (str | None)

name: str
version: str
title: str | None
app_type: str | None
description: str | None
created_at: str | None
updated_at: str | None
class kelvin.sdk.commands.apps.AppDeleteResult(app_name, version, deleted)[source]

Bases: object

Result of deleting an app.

Parameters:

  • app_name (str)

  • version (str | None)

  • deleted (bool)

app_name: str
version: str | None
deleted: bool
class kelvin.sdk.commands.apps.AppDownloadResult(app_name, version, image_name)[source]

Bases: object

Result of downloading an app.

Parameters:

  • app_name (str)

  • version (str)

  • image_name (str)

app_name: str
version: str
image_name: str
exception kelvin.sdk.commands.apps.AppsCommandsError(message=None, exit_code=None)[source]

Bases: CLIError

Base exception for apps command operations.

Parameters:

  • message (str)

  • exit_code (int)

exception kelvin.sdk.commands.apps.AppNotFoundError(app_name, version=None)[source]

Bases: AppsCommandsError

Raised when an app is not found.

Parameters:

  • app_name (str)

  • version (Optional[str])

exit_code: int = 66
app_name: str
version: str | None
exception kelvin.sdk.commands.apps.InvalidAppNameFormatError(app_name)[source]

Bases: AppsCommandsError

Raised when app name format is invalid.

Parameters:

app_name (str)

exit_code: int = 64
app_name: str
class kelvin.sdk.commands.apps.AppsCommands(client)[source]

Bases: object

Platform application registry commands.

Handles operations for managing applications on the platform’s registry. These are API-based operations that require authentication.

Parameters:

client (Client)

list(query=None)[source]

List all applications on the platform.

Parameters:

query (Optional[str]) – Optional search query to filter apps.

Return type:

AppListResult

Returns:

AppListResult containing list of apps.

search(query)[source]

Search for apps matching the query.

Parameters:

query (str) – Search query string.

Return type:

AppListResult

Returns:

AppListResult containing matching apps.

show(app_name)[source]

Show details of an application.

If app_name includes a version (e.g., “my-app:1.0.0”), shows version details. Otherwise, shows app details with all versions.

Parameters:

app_name (str) – Application name, optionally with version.

Return type:

Union[AppDetailResult, AppVersionDetailResult]

Returns:

AppDetailResult or AppVersionDetailResult with app details.

Raises:

AppNotFoundError – If app is not found.

delete(app_name)[source]

Delete an application or app version from the platform.

If app_name includes a version (e.g., “my-app:1.0.0”), deletes that version. Otherwise, deletes the entire app with all versions.

Parameters:

app_name (str) – Application name, optionally with version.

Return type:

AppDeleteResult

Returns:

AppDeleteResult indicating success.

Raises:

AppNotFoundError – If app is not found.

download(docker_service, registry_url, app_name, tag_local_name=True)[source]

Download an application from the platform.

Pulls the Docker image from the platform’s registry.

Parameters:

  • docker_service (DockerService) – Docker service for pulling images.

  • registry_url (str) – Docker registry URL.

  • app_name (str) – Application name with version (e.g., “my-app:1.0.0”).

  • tag_local_name (bool) – If True, tag image with local name (without registry).

Return type:

AppDownloadResult

Returns:

AppDownloadResult with download information.

Raises:

Auth Commands

Business logic for authentication workflows (login, logout, token management).

Authentication command business logic.

class kelvin.sdk.commands.auth.DockerLoginInfo(login_url, is_custom_registry=False, login_failed=False)[source]

Bases: object

Docker registry login status from auth flow.

Parameters:

  • login_url (str)

  • is_custom_registry (bool)

  • login_failed (bool)

login_url: str
is_custom_registry: bool = False
login_failed: bool = False
class kelvin.sdk.commands.auth.VersionInfo(current, recommended)[source]

Bases: object

CLI version compatibility info.

Parameters:

  • current (str)

  • recommended (str)

current: str
recommended: str
class kelvin.sdk.commands.auth.LoginResult(url, docker=None, version=None)[source]

Bases: object

Result of login operation.

Parameters:
url: str
docker: DockerLoginInfo | None = None
version: VersionInfo | None = None
class kelvin.sdk.commands.auth.TokenResult(access_token, expires_at, full_credentials=None)[source]

Bases: object

Result of token operation.

Parameters:
access_token: str
expires_at: int
full_credentials: dict[str, object] | None = None
exception kelvin.sdk.commands.auth.AuthCommandsError(message=None, exit_code=None)[source]

Bases: AuthenticationError

Base exception for auth command operations.

Parameters:

  • message (str)

  • exit_code (int)

exception kelvin.sdk.commands.auth.MissingSessionError(message=None, exit_code=None)[source]

Bases: AuthCommandsError

Raised when no active session is found.

Parameters:

  • message (str)

  • exit_code (int)

exception kelvin.sdk.commands.auth.MissingCredentialsError(message=None, exit_code=None)[source]

Bases: AuthCommandsError

Raised when credentials are missing for a session.

Parameters:

  • message (str)

  • exit_code (int)

exception kelvin.sdk.commands.auth.TokenRefreshError(message=None, exit_code=None)[source]

Bases: AuthCommandsError

Raised when token refresh is not possible.

Parameters:

  • message (str)

  • exit_code (int)

class kelvin.sdk.commands.auth.AuthCommands(auth, credentials, session, docker)[source]

Bases: object

Authentication command business logic.

Receives complete, validated inputs from CLI layer. Orchestrates multiple services to implement workflows.

Parameters:
auth: AuthService
credentials: CredentialStore
session: SessionService
docker: DockerService
login_browser(url, open_browser=True, on_auth_url=None)[source]

Browser SSO login workflow.

Return type:

LoginResult

Parameters:
login_password(url, username, password, totp=None)[source]

Password login workflow.

Return type:

LoginResult

Parameters:
login_with_stored_credentials(url)[source]

Attempt login using stored keyring credentials.

Checks for stored credentials, validates or refreshes tokens, and verifies against Keycloak. Returns None if stored credentials are unavailable or invalid, signaling the caller to prompt.

Return type:

Optional[LoginResult]

Parameters:

url (str)

reset_url(url)[source]

Clear stored credentials for a specific URL.

Return type:

None

Parameters:

url (str)

login_client_credentials(url, client_id, client_secret)[source]

Client credentials login workflow.

Return type:

LoginResult

Parameters:

  • url (str)

  • client_id (str)

  • client_secret (str)

logout()[source]

Logout workflow.

Return type:

None

get_token(full=False)[source]

Get current token, refreshing if needed.

Return type:

TokenResult

Parameters:

full (bool)

reset()[source]

Full reset of auth state.

Return type:

None

MLflow Commands

Business logic for MLflow-based application creation and model import.

MLflow command business logic.

This module contains the business logic for MLflow commands, orchestrating the MLflow service with the template service to create and manage MLflow-based Kelvin applications.

exception kelvin.sdk.commands.mlflow.MLflowCommandsError(message=None, exit_code=None)[source]

Bases: CLIError

Base exception for MLflow command errors.

Parameters:

  • message (str)

  • exit_code (int)

exit_code: int = 70
exception kelvin.sdk.commands.mlflow.AppDirectoryExistsError(app_path)[source]

Bases: MLflowCommandsError

Raised when app directory already exists.

Parameters:

app_path (Path)

Return type:

None

exit_code: int = 73
app_path: Path
exception kelvin.sdk.commands.mlflow.AppDirectoryNotFoundError(app_path)[source]

Bases: MLflowCommandsError

Raised when app directory doesn’t exist for import.

Parameters:

app_path (Path)

Return type:

None

exit_code: int = 66
app_path: Path
exception kelvin.sdk.commands.mlflow.MissingRequiredOptionError(option_name)[source]

Bases: MLflowCommandsError

Raised when a required option is missing in no-prompt mode.

Parameters:

option_name (str)

Return type:

None

exit_code: int = 64
option_name: str
exception kelvin.sdk.commands.mlflow.NoModelsFoundError(registry_uri)[source]

Bases: MLflowCommandsError

Raised when no models found in registry.

Parameters:

registry_uri (str)

Return type:

None

exit_code: int = 66
registry_uri: str
class kelvin.sdk.commands.mlflow.ModelInfo(name, description, latest_version)[source]

Bases: object

Model information from MLflow registry.

Encapsulates MLflow RegisteredModel details to avoid exposing MLflow internals to outer layers.

Parameters:

  • name (str)

  • description (str | None)

  • latest_version (str | None)

name: str
description: str | None
latest_version: str | None
class kelvin.sdk.commands.mlflow.MLflowListResult(models)[source]

Bases: object

Result of listing MLflow models.

Parameters:

models (list[ModelInfo])

models: list[ModelInfo]
class kelvin.sdk.commands.mlflow.MLflowCreateResult(app_name, app_path, files_created, model_inputs, model_outputs)[source]

Bases: object

Result of creating app from MLflow model.

Parameters:
app_name: str
app_path: Path
files_created: list[str]
model_inputs: list[dict[str, str]]
model_outputs: list[dict[str, str]]
class kelvin.sdk.commands.mlflow.MLflowImportResult(app_path, model_path, config_updated, model_inputs, model_outputs)[source]

Bases: object

Result of importing MLflow model into app.

Parameters:
app_path: Path
model_path: Path
config_updated: bool
model_inputs: list[dict[str, str]]
model_outputs: list[dict[str, str]]
class kelvin.sdk.commands.mlflow.MLflowCommands(mlflow_service, template_service)[source]

Bases: object

MLflow command business logic.

Orchestrates MLflow service operations with template service for creating and managing MLflow-based Kelvin applications.

Parameters:
__init__(mlflow_service, template_service)[source]

Initialize MLflowCommands.

Parameters:

  • mlflow_service (MLflowService) – Service for MLflow registry operations.

  • template_service (TemplateService) – Service for template operations.

Return type:

None

list_models(registry_uri)[source]

List models in MLflow registry.

Parameters:

registry_uri (str) – URI of the MLflow registry.

Return type:

MLflowListResult

Returns:

MLflowListResult with list of models.

Raises:
get_model_versions(registry_uri, model_name)[source]

Get versions of a model.

Parameters:

  • registry_uri (str) – URI of the MLflow registry.

  • model_name (str) – Name of the model.

Return type:

list[ModelVersion]

Returns:

List of model versions.

Raises:
create_app(registry_uri, model_name, model_version, app_name, app_dir=None, app_version='1.0.0', overwrite=False)[source]

Create a Kelvin app from an MLflow model.

Parameters:

  • registry_uri (str) – URI of the MLflow registry.

  • model_name (str) – Name of the model.

  • model_version (str) – Version of the model.

  • app_name (str) – Name for the new app.

  • app_dir (Optional[Path]) – Parent directory for the app (default: current directory).

  • app_version (str) – Version for the new app (default: “1.0.0”).

  • overwrite (bool) – If True, overwrite existing app directory.

Return type:

MLflowCreateResult

Returns:

MLflowCreateResult with created app information.

Raises:
import_model(registry_uri, model_name, model_version, app_dir, update_config=False, overwrite=False)[source]

Import an MLflow model into an existing Kelvin app.

Parameters:

  • registry_uri (str) – URI of the MLflow registry.

  • model_name (str) – Name of the model.

  • model_version (str) – Version of the model.

  • app_dir (Path) – Path to the existing app directory.

  • update_config (bool) – If True, update app.yaml with model inputs/outputs.

  • overwrite (bool) – If True, overwrite existing model directory.

Return type:

MLflowImportResult

Returns:

MLflowImportResult with import information.

Raises:
normalize_app_name(model_name)[source]

Normalize model name to a valid app name.

Converts to lowercase and replaces spaces/underscores with hyphens.

Parameters:

model_name (str) – The model name to normalize.

Return type:

str

Returns:

Normalized app name.

Secret Commands

Business logic for platform secret management operations.

Secret command business logic.

Platform secret management operations. These commands interact with the API to manage secrets on the platform.

class kelvin.sdk.commands.secret.SecretInfo(name, created_at, updated_at)[source]

Bases: object

Secret information.

Parameters:

  • name (str)

  • created_at (str | None)

  • updated_at (str | None)

name: str
created_at: str | None
updated_at: str | None
class kelvin.sdk.commands.secret.SecretListResult(secrets, total)[source]

Bases: object

Result of listing secrets.

Parameters:
secrets: list[SecretInfo]
total: int
class kelvin.sdk.commands.secret.SecretCreateResult(name, created)[source]

Bases: object

Result of creating a secret.

Parameters:
name: str
created: bool
class kelvin.sdk.commands.secret.SecretUpdateResult(name, updated)[source]

Bases: object

Result of updating a secret.

Parameters:
name: str
updated: bool
class kelvin.sdk.commands.secret.SecretDeleteResult(name, deleted)[source]

Bases: object

Result of deleting a secret.

Parameters:
name: str
deleted: bool
exception kelvin.sdk.commands.secret.SecretCommandsError(message=None, exit_code=None)[source]

Bases: CLIError

Base exception for secret command operations.

Parameters:

  • message (str)

  • exit_code (int)

exception kelvin.sdk.commands.secret.SecretNotFoundError(secret_name)[source]

Bases: SecretCommandsError

Raised when a secret is not found.

Parameters:

secret_name (str)

exit_code: int = 66
secret_name: str
exception kelvin.sdk.commands.secret.SecretAlreadyExistsError(secret_name)[source]

Bases: SecretCommandsError

Raised when a secret already exists.

Parameters:

secret_name (str)

exit_code: int = 65
secret_name: str
class kelvin.sdk.commands.secret.SecretCommands(client)[source]

Bases: object

Platform secret commands.

Handles operations for managing secrets on the platform. These are API-based operations that require authentication.

Parameters:

client (Client)

list(query=None)[source]

List all secrets on the platform.

Parameters:

query (Optional[str]) – Optional search query to filter secrets.

Return type:

SecretListResult

Returns:

SecretListResult containing list of secrets.

create(name, value)[source]

Create a new secret on the platform.

Parameters:

  • name (str) – The name of the secret.

  • value (str) – The value of the secret.

Return type:

SecretCreateResult

Returns:

SecretCreateResult indicating success.

Raises:

SecretAlreadyExistsError – If secret already exists.

update(name, value)[source]

Update an existing secret on the platform.

Parameters:

  • name (str) – The name of the secret to update.

  • value (str) – The new value for the secret.

Return type:

SecretUpdateResult

Returns:

SecretUpdateResult indicating success.

Raises:

SecretNotFoundError – If secret is not found.

delete(name)[source]

Delete a secret from the platform.

Parameters:

name (str) – The name of the secret to delete.

Return type:

SecretDeleteResult

Returns:

SecretDeleteResult indicating success.

Raises:

SecretNotFoundError – If secret is not found.

Workload Commands

Business logic for workload deployment and management operations.

Workload command business logic.

class kelvin.sdk.commands.workload.WorkloadInfo(name, title, cluster_name, app_name, app_version, status_state, status_message, last_seen)[source]

Bases: object

Summary info for a workload in list view.

Parameters:

  • name (str)

  • title (str | None)

  • cluster_name (str | None)

  • app_name (str | None)

  • app_version (str | None)

  • status_state (str | None)

  • status_message (str | None)

  • last_seen (datetime | None)

name: str
title: str | None
cluster_name: str | None
app_name: str | None
app_version: str | None
status_state: str | None
status_message: str | None
last_seen: datetime | None
class kelvin.sdk.commands.workload.WorkloadDetails(name, title, cluster_name, node_name, app_name, app_version, app_type, status_state, status_message, last_seen, download_status, download_error, created_at, created_by, updated_at, updated_by, runtime)[source]

Bases: object

Full workload details for show command.

Parameters:

  • name (str)

  • title (str | None)

  • cluster_name (str | None)

  • node_name (str | None)

  • app_name (str | None)

  • app_version (str | None)

  • app_type (str | None)

  • status_state (str | None)

  • status_message (str | None)

  • last_seen (datetime | None)

  • download_status (str | None)

  • download_error (str | None)

  • created_at (datetime | None)

  • created_by (str | None)

  • updated_at (datetime | None)

  • updated_by (str | None)

  • runtime (dict[str, object] | None)

name: str
title: str | None
cluster_name: str | None
node_name: str | None
app_name: str | None
app_version: str | None
app_type: str | None
status_state: str | None
status_message: str | None
last_seen: datetime | None
download_status: str | None
download_error: str | None
created_at: datetime | None
created_by: str | None
updated_at: datetime | None
updated_by: str | None
runtime: dict[str, object] | None
class kelvin.sdk.commands.workload.WorkloadListResult(workloads, total_count)[source]

Bases: object

Result of list workloads operation.

Parameters:
workloads: list[WorkloadInfo]
total_count: int
class kelvin.sdk.commands.workload.WorkloadLogsResult(workload_name, logs, has_logs)[source]

Bases: object

Result of get logs operation.

Parameters:
workload_name: str
logs: dict[str, list[str]]
has_logs: bool
class kelvin.sdk.commands.workload.DeployResult(workload_name, app_name, app_version, cluster_name, is_legacy)[source]

Bases: object

Result of deploy workload operation.

Parameters:

  • workload_name (str)

  • app_name (str)

  • app_version (str)

  • cluster_name (str | None)

  • is_legacy (bool)

workload_name: str
app_name: str
app_version: str
cluster_name: str | None
is_legacy: bool
class kelvin.sdk.commands.workload.UpdateResult(workload_name, app_name, app_version, is_legacy)[source]

Bases: object

Result of update workload operation.

Parameters:

  • workload_name (str)

  • app_name (str)

  • app_version (str)

  • is_legacy (bool)

workload_name: str
app_name: str
app_version: str
is_legacy: bool
exception kelvin.sdk.commands.workload.WorkloadCommandsError(message=None, exit_code=None)[source]

Bases: CLIError

Base exception for workload command operations.

Parameters:

  • message (str)

  • exit_code (int)

exception kelvin.sdk.commands.workload.WorkloadNotFoundError(workload_name)[source]

Bases: WorkloadCommandsError

Raised when workload is not found.

Parameters:

workload_name (str)

exit_code: int = 66
exception kelvin.sdk.commands.workload.AppConfigError(message=None, exit_code=None)[source]

Bases: WorkloadCommandsError

Raised when app configuration is invalid or cannot be parsed.

Parameters:

  • message (str)

  • exit_code (int)

exit_code: int = 65
exception kelvin.sdk.commands.workload.RuntimeConfigRequiredError[source]

Bases: WorkloadCommandsError

Raised when runtime config is required but not provided.

Return type:

None

exit_code: int = 64
exception kelvin.sdk.commands.workload.LegacyDeployMissingArgsError(missing_fields)[source]

Bases: WorkloadCommandsError

Raised when legacy deploy is missing required arguments.

Parameters:

missing_fields (list[str])

Return type:

None

exit_code: int = 64
exception kelvin.sdk.commands.workload.SchemaValidationFailedError(errors)[source]

Bases: WorkloadCommandsError

Raised when app config fails schema validation.

Parameters:

errors (list[str])

Return type:

None

exit_code: int = 65
validation_errors: list[str]
kelvin.sdk.commands.workload.get_status_category(state)[source]

Get the display category for a status state.

Return type:

str

Parameters:

state (str | None)

class kelvin.sdk.commands.workload.WorkloadCommands(client, schema_service)[source]

Bases: object

Workload command business logic.

Receives complete, validated inputs from CLI layer. Uses API client to perform operations.

Parameters:
list_workloads(*, search=None, app_name=None, node_name=None, cluster_name=None, status=None)[source]

List workloads with optional filters.

Return type:

WorkloadListResult

Parameters:

  • search (str | None)

  • app_name (str | None)

  • node_name (str | None)

  • cluster_name (str | None)

  • status (str | None)

show_workload(workload_name)[source]

Get detailed information about a workload.

Return type:

WorkloadDetails

Parameters:

workload_name (str)

start_workload(workload_name)[source]

Start a workload.

Return type:

None

Parameters:

workload_name (str)

stop_workload(workload_name)[source]

Stop a workload.

Return type:

None

Parameters:

workload_name (str)

undeploy_workload(workload_name)[source]

Undeploy (delete) a workload.

Return type:

None

Parameters:

workload_name (str)

get_logs(workload_name, *, tail_lines=None, since_time=None)[source]

Get workload logs.

Return type:

WorkloadLogsResult

Parameters:

  • workload_name (str)

  • tail_lines (int | None)

  • since_time (datetime | None)

follow_logs(workload_name, *, tail_lines=None, poll_interval=5.0)[source]

Follow workload logs, yielding new logs as they arrive.

This is a generator that polls for new logs.

Return type:

Iterator[WorkloadLogsResult]

Parameters:

  • workload_name (str)

  • tail_lines (int | None)

  • poll_interval (float)

deploy_workload(*, app_config_path, runtime_path=None, cluster_name=None, workload_name=None, workload_title=None)[source]

Deploy a new workload.

Parses the app configuration, determines if it is a legacy or new-style app, and creates the workload via the appropriate API.

Return type:

DeployResult

Parameters:

  • app_config_path (str)

  • runtime_path (str | None)

  • cluster_name (str | None)

  • workload_name (str | None)

  • workload_title (str | None)

update_workload(*, workload_name, app_config_path, runtime_path=None, workload_title=None)[source]

Update an existing workload.

Fetches the existing workload, parses the app configuration, and applies the update via the appropriate API.

Return type:

UpdateResult

Parameters:

  • workload_name (str)

  • app_config_path (str)

  • runtime_path (str | None)

  • workload_title (str | None)