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.
Important enumerations used throughout the ZenML API. Enums provide type-safe constants for statuses, component types, stages, and configuration options.
class ExecutionStatus(str, Enum):
"""
Pipeline and step execution status.
Values:
- INITIALIZING: Execution is initializing
- PROVISIONING: Resources being provisioned
- RUNNING: Currently executing
- COMPLETED: Successfully completed
- FAILED: Execution failed
- CACHED: Result retrieved from cache
- RETRYING: Being retried after failure
- RETRIED: Was retried
- STOPPED: Stopped by user
- STOPPING: In process of stopping
"""
INITIALIZING = "initializing"
PROVISIONING = "provisioning"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CACHED = "cached"
RETRYING = "retrying"
RETRIED = "retried"
STOPPED = "stopped"
STOPPING = "stopping"Import from:
from zenml.enums import ExecutionStatusclass StackComponentType(str, Enum):
"""
Stack component types.
Values:
- ORCHESTRATOR: Pipeline orchestration (required)
- ARTIFACT_STORE: Artifact storage (required)
- CONTAINER_REGISTRY: Container image registry
- IMAGE_BUILDER: Container image builder
- STEP_OPERATOR: Remote step execution
- EXPERIMENT_TRACKER: Experiment tracking
- MODEL_DEPLOYER: Model deployment
- FEATURE_STORE: Feature store
- MODEL_REGISTRY: Model registry
- DATA_VALIDATOR: Data validation
- ALERTER: Alerting/notifications
- ANNOTATOR: Data annotation
- DEPLOYER: Pipeline deployment
"""
ORCHESTRATOR = "orchestrator"
ARTIFACT_STORE = "artifact_store"
CONTAINER_REGISTRY = "container_registry"
IMAGE_BUILDER = "image_builder"
STEP_OPERATOR = "step_operator"
EXPERIMENT_TRACKER = "experiment_tracker"
MODEL_DEPLOYER = "model_deployer"
FEATURE_STORE = "feature_store"
MODEL_REGISTRY = "model_registry"
DATA_VALIDATOR = "data_validator"
ALERTER = "alerter"
ANNOTATOR = "annotator"
DEPLOYER = "deployer"Import from:
from zenml.enums import StackComponentTypeclass ModelStages(str, Enum):
"""
Model lifecycle stages.
Values:
- NONE: No specific stage
- STAGING: Model in staging environment
- PRODUCTION: Model in production
- ARCHIVED: Archived model
- LATEST: Latest model version (special marker)
"""
NONE = "none"
STAGING = "staging"
PRODUCTION = "production"
ARCHIVED = "archived"
LATEST = "latest"Import from:
from zenml.enums import ModelStagesclass ArtifactType(str, Enum):
"""
Types of artifacts.
Values:
- DATA: General data artifacts
- MODEL: Model artifacts
- DATA_ANALYSIS: Data analysis results
- SERVICE: Service artifacts
- STATISTICS: Statistical data
- SCHEMA: Schema definitions
- BASE: Base artifact type
"""
DATA = "data"
MODEL = "model"
DATA_ANALYSIS = "data_analysis"
SERVICE = "service"
STATISTICS = "statistics"
SCHEMA = "schema"
BASE = "base"Import from:
from zenml.enums import ArtifactTypeclass VisualizationType(str, Enum):
"""
Types of artifact visualizations.
Values:
- CSV: CSV data visualization
- HTML: HTML visualization
- IMAGE: Image visualization
- MARKDOWN: Markdown visualization
- JSON: JSON visualization
"""
CSV = "csv"
HTML = "html"
IMAGE = "image"
MARKDOWN = "markdown"
JSON = "json"Import from:
from zenml.enums import VisualizationTypeclass ExecutionMode(str, Enum):
"""
Pipeline execution mode for error handling.
Values:
- FAIL_FAST: Stop on first step failure
- STOP_ON_FAILURE: Deprecated, same as FAIL_FAST
- CONTINUE_ON_FAILURE: Continue executing independent steps
"""
FAIL_FAST = "fail_fast"
STOP_ON_FAILURE = "stop_on_failure"
CONTINUE_ON_FAILURE = "continue_on_failure"Import from:
from zenml.enums import ExecutionModeclass StoreType(str, Enum):
"""
ZenML store backend types.
Values:
- SQL: SQL database store
- REST: REST API store (ZenML server)
"""
SQL = "sql"
REST = "rest"Import from:
from zenml.enums import StoreTypeclass SecretsStoreType(str, Enum):
"""
Secrets store backend types.
Values:
- NONE: No secrets store
- SQL: SQL database
- AWS: AWS Secrets Manager
- GCP: Google Secret Manager
- AZURE: Azure Key Vault
- HASHICORP: HashiCorp Vault
- CUSTOM: Custom secrets store
"""
NONE = "none"
SQL = "sql"
AWS = "aws"
GCP = "gcp"
AZURE = "azure"
HASHICORP = "hashicorp"
CUSTOM = "custom"Import from:
from zenml.enums import SecretsStoreTypeclass LoggingLevels(str, Enum):
"""
Logging level options.
Values:
- NOTSET: Not set
- DEBUG: Debug messages
- INFO: Informational messages
- WARN: Warning messages
- WARNING: Warning messages (alias)
- ERROR: Error messages
- CRITICAL: Critical errors
"""
NOTSET = "NOTSET"
DEBUG = "DEBUG"
INFO = "INFO"
WARN = "WARN"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"Import from:
from zenml.enums import LoggingLevelsclass TaggableResourceTypes(str, Enum):
"""
Resource types that can be tagged.
Values:
- ARTIFACT: Artifact resources
- ARTIFACT_VERSION: Artifact version resources
- MODEL: Model resources
- MODEL_VERSION: Model version resources
- PIPELINE: Pipeline resources
- PIPELINE_RUN: Pipeline run resources
- RUN_TEMPLATE: Run template resources
- PIPELINE_SNAPSHOT: Pipeline snapshot resources
- DEPLOYMENT: Deployment resources
"""
ARTIFACT = "artifact"
ARTIFACT_VERSION = "artifact_version"
MODEL = "model"
MODEL_VERSION = "model_version"
PIPELINE = "pipeline"
PIPELINE_RUN = "pipeline_run"
RUN_TEMPLATE = "run_template"
PIPELINE_SNAPSHOT = "pipeline_snapshot"
DEPLOYMENT = "deployment"Import from:
from zenml.enums import TaggableResourceTypesclass MetadataResourceTypes(str, Enum):
"""
Resource types that can have metadata.
Values:
- PIPELINE_RUN: Pipeline run metadata
- STEP_RUN: Step run metadata
- ARTIFACT_VERSION: Artifact version metadata
- MODEL_VERSION: Model version metadata
- SCHEDULE: Schedule metadata
"""
PIPELINE_RUN = "pipeline_run"
STEP_RUN = "step_run"
ARTIFACT_VERSION = "artifact_version"
MODEL_VERSION = "model_version"
SCHEDULE = "schedule"Import from:
from zenml.enums import MetadataResourceTypesclass ContainerRegistryFlavor(str, Enum):
"""
Container registry flavors.
Values:
- DEFAULT: Generic/default container registry
- GITHUB: GitHub Container Registry
- DOCKERHUB: Docker Hub
- GCP: Google Container Registry
- AZURE: Azure Container Registry
"""
DEFAULT = "default"
GITHUB = "github"
DOCKERHUB = "dockerhub"
GCP = "gcp"
AZURE = "azure"Import from:
from zenml.enums import ContainerRegistryFlavorclass ColorVariants(str, Enum):
"""
Color options for tags and UI elements.
Values:
- GREY, PURPLE, RED, GREEN, YELLOW, ORANGE, LIME,
TEAL, TURQUOISE, MAGENTA, BLUE
"""
GREY = "grey"
PURPLE = "purple"
RED = "red"
GREEN = "green"
YELLOW = "yellow"
ORANGE = "orange"
LIME = "lime"
TEAL = "teal"
TURQUOISE = "turquoise"
MAGENTA = "magenta"
BLUE = "blue"Import from:
from zenml.enums import ColorVariantsclass ServiceState(str, Enum):
"""
Service states.
Values:
- INACTIVE: Service not running
- ACTIVE: Service running
- PENDING_STARTUP: Starting up
- PENDING_SHUTDOWN: Shutting down
- ERROR: Error state
- SCALED_TO_ZERO: Scaled to zero (serverless)
"""
INACTIVE = "inactive"
ACTIVE = "active"
PENDING_STARTUP = "pending_startup"
PENDING_SHUTDOWN = "pending_shutdown"
ERROR = "error"
SCALED_TO_ZERO = "scaled_to_zero"Import from:
from zenml.enums import ServiceStateclass EnvironmentType(str, Enum):
"""
Execution environment types.
Values include:
- NATIVE: Native Python environment
- CONTAINER: Docker container
- KUBERNETES: Kubernetes pod
- NOTEBOOK: Jupyter notebook
- COLAB: Google Colab
- GITHUB_ACTION: GitHub Actions
- GITLAB_CI: GitLab CI
- And many more CI/CD and cloud environments
"""
NATIVE = "native"
CONTAINER = "container"
KUBERNETES = "kubernetes"
NOTEBOOK = "notebook"
COLAB = "colab"
GITHUB_ACTION = "github_action"
GITLAB_CI = "gitlab_ci"
# ... and moreImport from:
from zenml.enums import EnvironmentTypefrom zenml.client import Client
from zenml.enums import ExecutionStatus
client = Client()
run = client.get_pipeline_run("run_id")
if run.status == ExecutionStatus.COMPLETED:
print("Pipeline completed successfully")
elif run.status == ExecutionStatus.FAILED:
print("Pipeline failed")
elif run.status == ExecutionStatus.RUNNING:
print("Pipeline still running")from zenml.client import Client
from zenml.enums import StackComponentType
client = Client()
# List all orchestrators
orchestrators = client.list_stack_components(
type=StackComponentType.ORCHESTRATOR
)
# List all artifact stores
stores = client.list_stack_components(
type=StackComponentType.ARTIFACT_STORE
)from zenml import Model
from zenml.enums import ModelStages
from zenml.client import Client
# Reference production model
production_model = Model(
name="my_model",
version=ModelStages.PRODUCTION
)
# Update model stage
client = Client()
client.update_model_version(
model_name_or_id="my_model",
version_name_or_id="1.0.0",
stage=ModelStages.PRODUCTION
)from zenml import save_artifact
from zenml.enums import ArtifactType
# Save with explicit type
model_artifact = save_artifact(
data={"weights": [0.1, 0.2]},
name="my_model",
artifact_type=ArtifactType.MODEL
)
data_artifact = save_artifact(
data=[1, 2, 3, 4, 5],
name="training_data",
artifact_type=ArtifactType.DATA
)from zenml.client import Client
from zenml.enums import ColorVariants
client = Client()
# Create colored tags
client.create_tag(name="production", color=ColorVariants.GREEN)
client.create_tag(name="experimental", color=ColorVariants.YELLOW)
client.create_tag(name="deprecated", color=ColorVariants.RED)from zenml import add_tags
from zenml.enums import TaggableResourceTypes
from zenml.client import Client
client = Client()
# Get resource ID
artifact = client.get_artifact("my_artifact")
# Add tags
add_tags(
resource_id=artifact.id,
resource_type=TaggableResourceTypes.ARTIFACT,
tags=["production", "validated"]
)from zenml.enums import LoggingLevels
import os
# Set logging level via environment
os.environ["ZENML_LOGGING_VERBOSITY"] = LoggingLevels.DEBUGfrom zenml import pipeline
from zenml.enums import ExecutionMode
@pipeline(execution_mode=ExecutionMode.CONTINUE_ON_FAILURE)
def fault_tolerant_pipeline():
"""Pipeline continues even if some steps fail."""
pass
@pipeline(execution_mode=ExecutionMode.FAIL_FAST)
def strict_pipeline():
"""Pipeline stops on first failure."""
passfrom zenml.client import Client
from zenml.enums import StoreType
client = Client()
if client.zen_store.type == StoreType.REST:
print("Connected to ZenML server")
elif client.zen_store.type == StoreType.SQL:
print("Using local SQL store")Install with Tessl CLI
npx tessl i tessl/pypi-zenml