tessl/pypi-zenml
ZenML is a unified MLOps framework that extends battle-tested machine learning operations principles to support the entire AI stack, from classical machine learning models to advanced AI agents.
client.mddocs/
Client
The Client class is the main programmatic interface for interacting with ZenML. It provides access to all ZenML resources including stacks, pipelines, artifacts, models, secrets, users, projects, and more. The Client follows a singleton pattern and maintains connection to the ZenML server or local store.
Capabilities
Client Class
class Client:
"""
Main interface for interacting with ZenML programmatically.
The Client provides methods for managing all ZenML resources and
follows a singleton pattern for consistent state management.
Singleton access:
```python
from zenml.client import Client
client = Client()
```
"""
@staticmethod
def get_instance():
"""
Get singleton client instance.
Returns:
Client: The singleton client instance
"""
@staticmethod
def initialize(root: str = None):
"""
Initialize ZenML repository.
Parameters:
- root: Root directory for ZenML repository (default: current directory)
Returns:
Client: Initialized client instance
"""
@staticmethod
def is_repository_directory(path: str = None) -> bool:
"""
Check if path is a ZenML repository.
Parameters:
- path: Path to check (default: current directory)
Returns:
bool: True if path is a ZenML repository
"""
@staticmethod
def find_repository(path: str = None) -> str:
"""
Find ZenML repository in path hierarchy.
Parameters:
- path: Starting path (default: current directory)
Returns:
str: Path to repository root
Raises:
RuntimeError: If no repository found
"""
@staticmethod
def is_inside_repository(path: str = None) -> bool:
"""
Check if path is inside a ZenML repository.
Parameters:
- path: Path to check (default: current directory)
Returns:
bool: True if inside a repository
"""
@property
def zen_store(self):
"""
Access to the underlying ZenML store.
Returns:
ZenStore: Store instance for direct store operations
"""
@property
def root(self) -> str:
"""
Root directory of ZenML repository.
Returns:
str: Repository root path
"""
@property
def config_directory(self) -> str:
"""
Configuration directory path.
Returns:
str: Config directory path
"""
def uses_local_configuration(self) -> bool:
"""
Check if using local configuration.
Returns:
bool: True if using local config
"""Import from:
from zenml.client import ClientActive Resource Properties
class Client:
@property
def active_stack(self):
"""
Get the active stack object.
Returns:
Stack: Active stack with all components initialized
"""
@property
def active_stack_model(self):
"""
Get the active stack model.
Returns:
StackResponse: Active stack model with configuration
"""
@property
def active_project(self):
"""
Get the active project.
Returns:
ProjectResponse: Active project information
"""
@property
def active_user(self):
"""
Get the active user.
Returns:
UserResponse: Active user information
"""
def set_active_project(self, project_name_or_id: str):
"""
Set the active project.
Parameters:
- project_name_or_id: Project name or UUID
"""
def activate_stack(self, stack_name_or_id: str):
"""
Set the active stack.
Parameters:
- stack_name_or_id: Stack name or UUID
"""User Management
class Client:
def create_user(
self,
name: str,
password: str = None,
is_admin: bool = False
):
"""
Create a new user.
Parameters:
- name: Username
- password: User password (optional)
- is_admin: Whether user has admin privileges
Returns:
UserResponse: Created user
"""
def get_user(
self,
name_or_id: str = None,
allow_name_prefix_match: bool = True
):
"""
Get user by name or ID.
Parameters:
- name_or_id: Username or UUID (current user if None)
- allow_name_prefix_match: Allow prefix matching
Returns:
UserResponse: User information
"""
def list_users(
self,
name: str = None,
is_admin: bool = None,
**kwargs
):
"""
List all users.
Parameters:
- name: Filter by name
- is_admin: Filter by admin status
- **kwargs: Additional filter parameters
Returns:
Page[UserResponse]: Paginated list of users
"""
def update_user(
self,
name_or_id: str,
updated_name: str = None,
updated_full_name: str = None,
updated_email: str = None,
updated_email_opted_in: bool = None,
is_admin: bool = None
):
"""
Update user information.
Parameters:
- name_or_id: User name or UUID
- updated_name: New username
- updated_full_name: New full name
- updated_email: New email
- updated_email_opted_in: Email opt-in status
- is_admin: Admin status
Returns:
UserResponse: Updated user
"""
def deactivate_user(self, name_or_id: str):
"""
Deactivate a user.
Parameters:
- name_or_id: User name or UUID
"""
def delete_user(self, name_or_id: str):
"""
Delete a user.
Parameters:
- name_or_id: User name or UUID
"""Project Management
class Client:
def create_project(
self,
name: str,
description: str = None
):
"""
Create a new project.
Parameters:
- name: Project name
- description: Project description
Returns:
ProjectResponse: Created project
"""
def get_project(self, name_or_id: str = None):
"""
Get project by name or ID.
Parameters:
- name_or_id: Project name or UUID (active project if None)
Returns:
ProjectResponse: Project information
"""
def list_projects(self, name: str = None, **kwargs):
"""
List all projects.
Parameters:
- name: Filter by name
- **kwargs: Additional filter parameters
Returns:
Page[ProjectResponse]: Paginated list of projects
"""
def update_project(
self,
name_or_id: str,
updated_name: str = None,
updated_description: str = None
):
"""
Update project information.
Parameters:
- name_or_id: Project name or UUID
- updated_name: New project name
- updated_description: New description
Returns:
ProjectResponse: Updated project
"""
def delete_project(self, name_or_id: str):
"""
Delete a project.
Parameters:
- name_or_id: Project name or UUID
"""Stack Management
class Client:
def create_stack(
self,
name: str,
components: dict,
description: str = None
):
"""
Create a new stack.
Parameters:
- name: Stack name
- components: Dict mapping component type to component name/ID
- description: Stack description
Returns:
StackResponse: Created stack
Example:
```python
stack = client.create_stack(
name="my_stack",
components={
"orchestrator": "local_orchestrator",
"artifact_store": "local_artifact_store"
},
description="Local development stack"
)
```
"""
def get_stack(self, name_or_id: str = None):
"""
Get stack by name or ID.
Parameters:
- name_or_id: Stack name or UUID (active stack if None)
Returns:
StackResponse: Stack information
"""
def list_stacks(
self,
name: str = None,
component_id: str = None,
**kwargs
):
"""
List all stacks.
Parameters:
- name: Filter by name
- component_id: Filter by component UUID
- **kwargs: Additional filter parameters
Returns:
Page[StackResponse]: Paginated list of stacks
"""
def update_stack(
self,
name_or_id: str,
updated_name: str = None,
updated_description: str = None,
components: dict = None
):
"""
Update stack configuration.
Parameters:
- name_or_id: Stack name or UUID
- updated_name: New stack name
- updated_description: New description
- components: Updated components dict
Returns:
StackResponse: Updated stack
"""
def delete_stack(self, name_or_id: str):
"""
Delete a stack.
Parameters:
- name_or_id: Stack name or UUID
"""Stack Component Management
class Client:
def create_stack_component(
self,
name: str,
flavor: str,
component_type: str,
configuration: dict,
connector: str = None
):
"""
Create a new stack component.
Parameters:
- name: Component name
- flavor: Flavor name
- component_type: Component type (from StackComponentType enum)
- configuration: Component configuration dict
- connector: Service connector name/ID (optional)
Returns:
ComponentResponse: Created component
"""
def get_stack_component(
self,
name_or_id: str,
component_type: str = None
):
"""
Get stack component by name/ID/type.
Parameters:
- name_or_id: Component name or UUID
- component_type: Component type (from StackComponentType enum)
Returns:
ComponentResponse: Component information
"""
def list_stack_components(
self,
name: str = None,
type: str = None,
flavor_name: str = None,
**kwargs
):
"""
List stack components.
Parameters:
- name: Filter by name
- type: Filter by component type
- flavor_name: Filter by flavor
- **kwargs: Additional filter parameters
Returns:
Page[ComponentResponse]: Paginated list of components
"""
def update_stack_component(
self,
name_or_id: str,
component_type: str = None,
updated_name: str = None,
updated_configuration: dict = None,
updated_connector: str = None
):
"""
Update stack component.
Parameters:
- name_or_id: Component name or UUID
- component_type: Component type
- updated_name: New component name
- updated_configuration: Updated configuration
- updated_connector: Updated connector
Returns:
ComponentResponse: Updated component
"""
def delete_stack_component(
self,
name_or_id: str,
component_type: str = None
):
"""
Delete stack component.
Parameters:
- name_or_id: Component name or UUID
- component_type: Component type
"""Flavor Management
class Client:
def create_flavor(
self,
source: str,
component_type: str
):
"""
Register a custom flavor.
Parameters:
- source: Python source path (e.g., "my_module.MyFlavor")
- component_type: Component type (from StackComponentType enum)
Returns:
FlavorResponse: Created flavor
"""
def get_flavor(self, name_or_id: str):
"""
Get flavor by name or ID.
Parameters:
- name_or_id: Flavor name or UUID
Returns:
FlavorResponse: Flavor information
"""
def list_flavors(
self,
name: str = None,
type: str = None,
**kwargs
):
"""
List all flavors.
Parameters:
- name: Filter by name
- type: Filter by component type
- **kwargs: Additional filter parameters
Returns:
Page[FlavorResponse]: Paginated list of flavors
"""
def delete_flavor(self, name_or_id: str):
"""
Delete a flavor.
Parameters:
- name_or_id: Flavor name or UUID
"""
def get_flavors_by_type(self, component_type: str):
"""
Get flavors for a component type.
Parameters:
- component_type: Component type (from StackComponentType enum)
Returns:
Page[FlavorResponse]: List of flavors for the type
"""
def get_flavor_by_name_and_type(
self,
name: str,
component_type: str
):
"""
Get specific flavor by name and type.
Parameters:
- name: Flavor name
- component_type: Component type
Returns:
FlavorResponse: Flavor information
"""Pipeline Management
class Client:
def list_pipelines(
self,
name: str = None,
**kwargs
):
"""
List all pipelines.
Parameters:
- name: Filter by name
- **kwargs: Additional filter parameters
Returns:
Page[PipelineResponse]: Paginated list of pipelines
"""
def get_pipeline(self, name_or_id: str):
"""
Get pipeline by name or ID.
Parameters:
- name_or_id: Pipeline name or UUID
Returns:
PipelineResponse: Pipeline information
"""
def delete_pipeline(self, name_or_id: str):
"""
Delete a pipeline.
Parameters:
- name_or_id: Pipeline name or UUID
"""
def trigger_pipeline(
self,
name_or_id: str,
run_config: dict = None,
**kwargs
):
"""
Trigger a pipeline run.
Parameters:
- name_or_id: Pipeline name or UUID
- run_config: Run configuration dict
- **kwargs: Additional parameters
Returns:
PipelineRunResponse: Created pipeline run
"""Pipeline Run Management
class Client:
def get_pipeline_run(self, name_or_id: str):
"""
Get pipeline run by ID.
Parameters:
- name_or_id: Pipeline run UUID or name
Returns:
PipelineRunResponse: Pipeline run details
"""
def list_pipeline_runs(
self,
pipeline_id: str = None,
name: str = None,
status: str = None,
**kwargs
):
"""
List pipeline runs.
Parameters:
- pipeline_id: Filter by pipeline UUID
- name: Filter by run name
- status: Filter by execution status
- **kwargs: Additional filter parameters
Returns:
Page[PipelineRunResponse]: Paginated list of runs
"""
def delete_pipeline_run(self, name_or_id: str):
"""
Delete a pipeline run.
Parameters:
- name_or_id: Pipeline run UUID or name
"""Step Run Management
class Client:
def get_run_step(self, step_run_id: str):
"""
Get step run by ID.
Parameters:
- step_run_id: Step run UUID
Returns:
StepRunResponse: Step run details
"""
def list_run_steps(
self,
pipeline_run_id: str = None,
name: str = None,
**kwargs
):
"""
List step runs.
Parameters:
- pipeline_run_id: Filter by pipeline run UUID
- name: Filter by step name
- **kwargs: Additional filter parameters
Returns:
Page[StepRunResponse]: Paginated list of step runs
"""Artifact Management
class Client:
def get_artifact(self, name_or_id: str):
"""
Get artifact by name or ID.
Parameters:
- name_or_id: Artifact name or UUID
Returns:
ArtifactResponse: Artifact information
"""
def list_artifacts(
self,
name: str = None,
tag: str = None,
**kwargs
):
"""
List artifacts.
Parameters:
- name: Filter by name
- tag: Filter by tag
- **kwargs: Additional filter parameters
Returns:
Page[ArtifactResponse]: Paginated list of artifacts
"""
def update_artifact(
self,
name_or_id: str,
updated_name: str = None,
add_tags: list = None,
remove_tags: list = None
):
"""
Update artifact.
Parameters:
- name_or_id: Artifact name or UUID
- updated_name: New artifact name
- add_tags: Tags to add
- remove_tags: Tags to remove
Returns:
ArtifactResponse: Updated artifact
"""
def delete_artifact(self, name_or_id: str):
"""
Delete artifact.
Parameters:
- name_or_id: Artifact name or UUID
"""
def prune_artifacts(self, only_versions: bool = True):
"""
Prune unused artifacts.
Parameters:
- only_versions: Only delete unused versions (default: True)
Returns:
int: Number of artifacts/versions deleted
"""Artifact Version Management
class Client:
def get_artifact_version(
self,
name_or_id: str,
version: str = None
):
"""
Get artifact version.
Parameters:
- name_or_id: Artifact name or version UUID
- version: Version identifier (latest if None)
Returns:
ArtifactVersionResponse: Artifact version details
"""
def list_artifact_versions(
self,
name: str = None,
artifact_id: str = None,
**kwargs
):
"""
List artifact versions.
Parameters:
- name: Filter by artifact name
- artifact_id: Filter by artifact UUID
- **kwargs: Additional filter parameters
Returns:
Page[ArtifactVersionResponse]: Paginated list of versions
"""
def update_artifact_version(
self,
name_or_id: str,
version: str = None,
add_tags: list = None,
remove_tags: list = None
):
"""
Update artifact version.
Parameters:
- name_or_id: Artifact name or version UUID
- version: Version identifier
- add_tags: Tags to add
- remove_tags: Tags to remove
Returns:
ArtifactVersionResponse: Updated version
"""
def delete_artifact_version(
self,
name_or_id: str,
version: str = None
):
"""
Delete artifact version.
Parameters:
- name_or_id: Artifact name or version UUID
- version: Version identifier
"""Model Management
class Client:
def create_model(
self,
name: str,
license: str = None,
description: str = None,
tags: list = None
):
"""
Create model namespace.
Parameters:
- name: Model name
- license: License identifier
- description: Model description
- tags: List of tags
Returns:
ModelResponse: Created model
"""
def get_model(self, name_or_id: str):
"""
Get model by name or ID.
Parameters:
- name_or_id: Model name or UUID
Returns:
ModelResponse: Model information
"""
def list_models(
self,
name: str = None,
tag: str = None,
**kwargs
):
"""
List models.
Parameters:
- name: Filter by name
- tag: Filter by tag
- **kwargs: Additional filter parameters
Returns:
Page[ModelResponse]: Paginated list of models
"""
def update_model(
self,
name_or_id: str,
updated_name: str = None,
updated_license: str = None,
updated_description: str = None,
add_tags: list = None,
remove_tags: list = None
):
"""
Update model.
Parameters:
- name_or_id: Model name or UUID
- updated_name: New model name
- updated_license: New license
- updated_description: New description
- add_tags: Tags to add
- remove_tags: Tags to remove
Returns:
ModelResponse: Updated model
"""
def delete_model(self, name_or_id: str):
"""
Delete model.
Parameters:
- name_or_id: Model name or UUID
"""Model Version Management
class Client:
def create_model_version(
self,
model_name_or_id: str,
version: str = None,
description: str = None,
tags: list = None
):
"""
Create model version.
Parameters:
- model_name_or_id: Model name or UUID
- version: Version identifier
- description: Version description
- tags: List of tags
Returns:
ModelVersionResponse: Created model version
"""
def get_model_version(
self,
model_name_or_id: str,
version: str = None
):
"""
Get model version.
Parameters:
- model_name_or_id: Model name or UUID
- version: Version identifier or stage name
Returns:
ModelVersionResponse: Model version details
"""
def list_model_versions(
self,
model_name_or_id: str = None,
stage: str = None,
**kwargs
):
"""
List model versions.
Parameters:
- model_name_or_id: Filter by model name or UUID
- stage: Filter by stage
- **kwargs: Additional filter parameters
Returns:
Page[ModelVersionResponse]: Paginated list of versions
"""
def update_model_version(
self,
model_name_or_id: str,
version_name_or_id: str,
stage: str = None,
description: str = None,
add_tags: list = None,
remove_tags: list = None
):
"""
Update model version.
Parameters:
- model_name_or_id: Model name or UUID
- version_name_or_id: Version name or UUID
- stage: New stage
- description: New description
- add_tags: Tags to add
- remove_tags: Tags to remove
Returns:
ModelVersionResponse: Updated version
"""
def delete_model_version(
self,
model_name_or_id: str,
version_name_or_id: str
):
"""
Delete model version.
Parameters:
- model_name_or_id: Model name or UUID
- version_name_or_id: Version name or UUID
"""
def list_model_version_artifact_links(
self,
model_version_id: str
):
"""
List artifact links for model version.
Parameters:
- model_version_id: Model version UUID
Returns:
Page[ModelVersionArtifactResponse]: List of artifact links
"""
def delete_model_version_artifact_link(
self,
model_version_id: str,
artifact_version_id: str
):
"""
Delete artifact link from model version.
Parameters:
- model_version_id: Model version UUID
- artifact_version_id: Artifact version UUID
"""
def list_model_version_pipeline_run_links(
self,
model_version_id: str
):
"""
List pipeline run links for model version.
Parameters:
- model_version_id: Model version UUID
Returns:
Page[ModelVersionPipelineRunResponse]: List of pipeline run links
"""Secret Management
class Client:
def create_secret(
self,
name: str,
values: dict,
scope: str = "project"
):
"""
Create a new secret.
Parameters:
- name: Secret name
- values: Secret values dict
- scope: Secret scope ("project" or "user")
Returns:
SecretResponse: Created secret
"""
def get_secret(self, name_or_id: str):
"""
Get secret by name or ID.
Parameters:
- name_or_id: Secret name or UUID
Returns:
SecretResponse: Secret information (values not included)
"""
def list_secrets(
self,
name: str = None,
scope: str = None,
**kwargs
):
"""
List secrets.
Parameters:
- name: Filter by name
- scope: Filter by scope
- **kwargs: Additional filter parameters
Returns:
Page[SecretResponse]: Paginated list of secrets
"""
def update_secret(
self,
name_or_id: str,
updated_name: str = None,
updated_values: dict = None,
updated_scope: str = None
):
"""
Update secret.
Parameters:
- name_or_id: Secret name or UUID
- updated_name: New secret name
- updated_values: Updated values
- updated_scope: Updated scope
Returns:
SecretResponse: Updated secret
"""
def delete_secret(self, name_or_id: str):
"""
Delete secret.
Parameters:
- name_or_id: Secret name or UUID
"""
def backup_secrets(
self,
ignore_errors: bool = True,
delete_secrets: bool = False
):
"""
Backup secrets.
Parameters:
- ignore_errors: Continue on errors
- delete_secrets: Delete after backup
Returns:
dict: Backup data
"""
def restore_secrets(
self,
backup: dict,
ignore_errors: bool = False
):
"""
Restore secrets from backup.
Parameters:
- backup: Backup data dict
- ignore_errors: Continue on errors
"""Service Connector Management
class Client:
def create_service_connector(
self,
name: str,
type: str,
auth_method: str,
configuration: dict,
resource_types: list = None
):
"""
Create service connector.
Parameters:
- name: Connector name
- type: Connector type
- auth_method: Authentication method
- configuration: Configuration dict
- resource_types: List of resource types
Returns:
ServiceConnectorResponse: Created connector
"""
def get_service_connector(self, name_or_id: str):
"""
Get service connector.
Parameters:
- name_or_id: Connector name or UUID
Returns:
ServiceConnectorResponse: Connector information
"""
def list_service_connectors(
self,
name: str = None,
connector_type: str = None,
**kwargs
):
"""
List service connectors.
Parameters:
- name: Filter by name
- connector_type: Filter by type
- **kwargs: Additional filter parameters
Returns:
Page[ServiceConnectorResponse]: Paginated list of connectors
"""
def update_service_connector(
self,
name_or_id: str,
updated_name: str = None,
updated_configuration: dict = None
):
"""
Update service connector.
Parameters:
- name_or_id: Connector name or UUID
- updated_name: New connector name
- updated_configuration: Updated configuration
Returns:
ServiceConnectorResponse: Updated connector
"""
def delete_service_connector(self, name_or_id: str):
"""
Delete service connector.
Parameters:
- name_or_id: Connector name or UUID
"""
def verify_service_connector(
self,
name_or_id: str,
resource_type: str = None,
resource_id: str = None
):
"""
Verify connector credentials.
Parameters:
- name_or_id: Connector name or UUID
- resource_type: Resource type to verify
- resource_id: Resource ID to verify
Returns:
ServiceConnectorResourcesModel: Verification results
"""
def get_service_connector_client(
self,
name_or_id: str,
resource_type: str = None,
resource_id: str = None
):
"""
Get connector client.
Parameters:
- name_or_id: Connector name or UUID
- resource_type: Resource type
- resource_id: Resource ID
Returns:
Client object for the connector
"""
def list_service_connector_resources(
self,
connector_type: str = None,
resource_type: str = None
):
"""
List connector resources.
Parameters:
- connector_type: Filter by connector type
- resource_type: Filter by resource type
Returns:
list: List of available resources
"""
def list_service_connector_types(self):
"""
List connector types.
Returns:
list[ServiceConnectorTypeModel]: Available connector types
"""
def get_service_connector_type(self, connector_type: str):
"""
Get connector type.
Parameters:
- connector_type: Connector type identifier
Returns:
ServiceConnectorTypeModel: Connector type details
"""Tag Management
class Client:
def create_tag(
self,
name: str,
color: str = None
):
"""
Create tag.
Parameters:
- name: Tag name
- color: Tag color (from ColorVariants enum)
Returns:
TagResponse: Created tag
"""
def get_tag(self, name_or_id: str):
"""
Get tag.
Parameters:
- name_or_id: Tag name or UUID
Returns:
TagResponse: Tag information
"""
def list_tags(
self,
name: str = None,
**kwargs
):
"""
List tags.
Parameters:
- name: Filter by name
- **kwargs: Additional filter parameters
Returns:
Page[TagResponse]: Paginated list of tags
"""
def update_tag(
self,
name_or_id: str,
updated_name: str = None,
updated_color: str = None
):
"""
Update tag.
Parameters:
- name_or_id: Tag name or UUID
- updated_name: New tag name
- updated_color: New color
Returns:
TagResponse: Updated tag
"""
def delete_tag(self, name_or_id: str):
"""
Delete tag.
Parameters:
- name_or_id: Tag name or UUID
"""Additional Client Methods
The Client class includes 100+ methods total. Additional methods include:
- Code Repository Management:
create_code_repository(),get_code_repository(),list_code_repositories(),update_code_repository(),delete_code_repository() - Build Management:
get_build(),list_builds(),delete_build() - Schedule Management:
get_schedule(),list_schedules(),update_schedule(),delete_schedule() - Deployment Management:
get_deployment(),list_deployments(),provision_deployment(),deprovision_deployment(),delete_deployment(),get_deployment_logs() - Run Template Management:
create_run_template(),get_run_template(),list_run_templates(),update_run_template(),delete_run_template() - Service Management:
create_service(),get_service(),list_services(),update_service(),delete_service() - Event Source Management:
create_event_source(),get_event_source(),list_event_sources(),update_event_source(),delete_event_source() - Action Management:
create_action(),get_action(),list_actions(),update_action(),delete_action() - Trigger Management:
create_trigger(),get_trigger(),list_triggers(),update_trigger(),delete_trigger() - Trigger Execution:
get_trigger_execution(),list_trigger_executions(),delete_trigger_execution() - Service Account Management:
create_service_account(),get_service_account(),list_service_accounts(),update_service_account(),delete_service_account() - API Key Management:
create_api_key(),get_api_key(),list_api_keys(),update_api_key(),rotate_api_key(),delete_api_key(),set_api_key() - Device Authorization:
list_authorized_devices(),get_authorized_device(),update_authorized_device(),delete_authorized_device() - Server Settings:
get_settings(),update_server_settings() - Metadata:
create_run_metadata()
Usage Examples
Basic Client Usage
from zenml.client import Client
# Get client instance (singleton)
client = Client()
# Access active resources
print(f"Active stack: {client.active_stack_model.name}")
print(f"Active project: {client.active_project.name}")
print(f"Active user: {client.active_user.name}")
# Initialize a new repository
Client.initialize(root="/path/to/project")Managing Stacks
from zenml.client import Client
client = Client()
# Create a stack
stack = client.create_stack(
name="aws_stack",
components={
"orchestrator": "sagemaker_orchestrator",
"artifact_store": "s3_artifact_store",
"container_registry": "ecr_registry"
},
description="AWS production stack"
)
# List all stacks
stacks = client.list_stacks()
for s in stacks:
print(f"Stack: {s.name}")
# Switch active stack
client.activate_stack("aws_stack")
# Update stack
client.update_stack(
name_or_id="aws_stack",
updated_description="Updated AWS stack"
)Working with Pipeline Runs
from zenml.client import Client
client = Client()
# Get latest pipeline run
runs = client.list_pipeline_runs(
pipeline_id=client.get_pipeline("training_pipeline").id,
sort_by="created",
size=1
)
latest_run = runs.items[0]
# Get run details
print(f"Run: {latest_run.name}")
print(f"Status: {latest_run.status}")
print(f"Start time: {latest_run.start_time}")
# List steps in the run
steps = client.list_run_steps(pipeline_run_id=latest_run.id)
for step in steps:
print(f"Step: {step.name}, Status: {step.status}")
# Get step outputs
step = steps.items[0]
outputs = step.outputs
for name, artifact in outputs.items():
print(f"Output '{name}': {artifact.uri}")Managing Models
from zenml.client import Client
from zenml.enums import ModelStages
client = Client()
# Create model
model = client.create_model(
name="fraud_detector",
license="MIT",
description="Fraud detection model",
tags=["fraud", "classification"]
)
# Create version
version = client.create_model_version(
model_name_or_id=model.id,
version="1.0.0",
description="Initial release"
)
# Promote to production
client.update_model_version(
model_name_or_id=model.id,
version_name_or_id=version.id,
stage=ModelStages.PRODUCTION
)
# Get production version
prod_version = client.get_model_version(
model_name_or_id="fraud_detector",
version=ModelStages.PRODUCTION
)
print(f"Production version: {prod_version.version}")Managing Secrets
from zenml.client import Client
client = Client()
# Create secret
secret = client.create_secret(
name="aws_credentials",
values={
"aws_access_key_id": "AKIA...",
"aws_secret_access_key": "secret..."
},
scope="project"
)
# List secrets (values not returned)
secrets = client.list_secrets()
for s in secrets:
print(f"Secret: {s.name}")
# Update secret
client.update_secret(
name_or_id="aws_credentials",
updated_values={
"aws_access_key_id": "AKIA_NEW...",
"aws_secret_access_key": "new_secret..."
}
)
# Backup and restore
backup = client.backup_secrets()
# ... later ...
client.restore_secrets(backup)Install with Tessl CLI
npx tessl i tessl/pypi-zenml