CtrlK
BlogDocsLog inGet started
Tessl Logo

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.

Overview
Eval results
Files

enums.mddocs/

Enums

Important enumerations used throughout the ZenML API. Enums provide type-safe constants for statuses, component types, stages, and configuration options.

Capabilities

Execution Status

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 ExecutionStatus

Stack Component Type

class 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 StackComponentType

Model Stages

class 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 ModelStages

Artifact Type

class 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 ArtifactType

Visualization Type

class 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 VisualizationType

Execution Mode

class 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 ExecutionMode

Store Type

class 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 StoreType

Secrets Store Type

class 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 SecretsStoreType

Logging Levels

class 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 LoggingLevels

Taggable Resource Types

class 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 TaggableResourceTypes

Metadata Resource Types

class 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 MetadataResourceTypes

Container Registry Flavor

class 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 ContainerRegistryFlavor

Color Variants

class 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 ColorVariants

Service State

class 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 ServiceState

Environment Type

class 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 more

Import from:

from zenml.enums import EnvironmentType

Usage Examples

Checking Execution Status

from 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")

Filtering by Component Type

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
)

Using Model Stages

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
)

Artifact Type Specification

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
)

Tag Colors

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)

Tagging Resources

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"]
)

Logging Levels

from zenml.enums import LoggingLevels
import os

# Set logging level via environment
os.environ["ZENML_LOGGING_VERBOSITY"] = LoggingLevels.DEBUG

Execution Mode

from 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."""
    pass

Store Type Checking

from 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

docs

artifact-config.md

artifacts.md

client.md

config.md

enums.md

exceptions.md

hooks.md

index.md

integrations.md

materializers.md

metadata-tags.md

models.md

pipelines-and-steps.md

pydantic-models.md

services.md

stack-components.md

stacks.md

types.md

utilities.md

tile.json