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

Pydantic Models

Name: tessl/pypi-zenml
Author: tessl

ZenML uses Pydantic for data validation and API request/response modeling. The zenml.models module contains 200+ Pydantic model classes representing all ZenML resources and operations.

Model Categories

Base Models

Foundation classes for all ZenML Pydantic models.

class BaseZenModel:
    """Base Pydantic model for all ZenML models."""

class BaseRequest:
    """Base for request models."""

class BaseResponse:
    """Base for response models."""

class BaseUpdate:
    """Base for update models."""

class BaseFilter:
    """Base filter class for querying resources."""

class Page:
    """
    Paginated response container.

    Attributes:
    - index: Current page index
    - max_size: Maximum items per page
    - total_pages: Total number of pages
    - total: Total number of items
    - items: List of items on current page
    """

Import from:

from zenml.models import (
    BaseZenModel,
    BaseRequest,
    BaseResponse,
    Page
)

Resource Models

Each ZenML resource has multiple model variants:

Request: For creating resources
Response: Returned from API calls
Update: For updating resources
Filter: For querying/filtering resources

User Models

class UserRequest:
    """Request model for creating users."""

class UserResponse:
    """Response model with user information."""

class UserUpdate:
    """Update model for modifying users."""

class UserFilter:
    """Filter for querying users."""

Project Models

class ProjectRequest:
    """Request model for creating projects."""

class ProjectResponse:
    """Response model with project information."""

class ProjectUpdate:
    """Update model for modifying projects."""

class ProjectFilter:
    """Filter for querying projects."""

Stack Models

class StackRequest:
    """Request model for creating stacks."""

class StackResponse:
    """Response model with stack information."""

class StackUpdate:
    """Update model for modifying stacks."""

class StackFilter:
    """Filter for querying stacks."""

Pipeline Models

class PipelineRequest:
    """Request model for creating pipelines."""

class PipelineResponse:
    """Response model with pipeline information."""

class PipelineFilter:
    """Filter for querying pipelines."""

Pipeline Run Models

class PipelineRunRequest:
    """Request model for creating pipeline runs."""

class PipelineRunResponse:
    """
    Response model with pipeline run information.

    Attributes:
    - id: Run UUID
    - name: Run name
    - status: Execution status
    - pipeline: Associated pipeline
    - stack: Stack used for execution
    - user: User who created the run
    - start_time: Run start time
    - end_time: Run end time
    - metadata: Run metadata
    """

class PipelineRunFilter:
    """Filter for querying pipeline runs."""

Step Run Models

class StepRunRequest:
    """Request model for creating step runs."""

class StepRunResponse:
    """
    Response model with step run information.

    Attributes:
    - id: Step run UUID
    - name: Step name
    - status: Execution status
    - inputs: Input artifacts
    - outputs: Output artifacts
    - metadata: Step metadata
    - start_time: Step start time
    - end_time: Step end time
    """

class StepRunFilter:
    """Filter for querying step runs."""

Artifact Models

class ArtifactRequest:
    """Request model for creating artifacts."""

class ArtifactResponse:
    """Response model with artifact information."""

class ArtifactUpdate:
    """Update model for modifying artifacts."""

class ArtifactFilter:
    """Filter for querying artifacts."""

Artifact Version Models

class ArtifactVersionRequest:
    """Request model for creating artifact versions."""

class ArtifactVersionResponse:
    """
    Response model with artifact version information.

    Attributes:
    - id: Artifact version UUID
    - artifact: Parent artifact
    - version: Version identifier
    - uri: Storage URI
    - type: Artifact type
    - materializer: Materializer used
    - data_type: Python data type
    - tags: Associated tags
    - metadata: Version metadata
    """

class LazyArtifactVersionResponse:
    """Lazy-loaded artifact version response."""

class ArtifactVersionFilter:
    """Filter for querying artifact versions."""

Model Models

class ModelRequest:
    """Request model for creating models."""

class ModelResponse:
    """Response model with model information."""

class ModelUpdate:
    """Update model for modifying models."""

class ModelFilter:
    """Filter for querying models."""

Model Version Models

class ModelVersionRequest:
    """Request model for creating model versions."""

class ModelVersionResponse:
    """
    Response model with model version information.

    Attributes:
    - id: Model version UUID
    - model: Parent model
    - version: Version identifier
    - stage: Model stage (staging, production, etc.)
    - description: Version description
    - tags: Associated tags
    - metadata: Version metadata
    - artifact_links: Linked artifacts
    - pipeline_run_links: Linked pipeline runs
    """

class ModelVersionUpdate:
    """Update model for modifying model versions."""

class ModelVersionFilter:
    """Filter for querying model versions."""

Secret Models

class SecretRequest:
    """Request model for creating secrets."""

class SecretResponse:
    """Response model with secret information (no values)."""

class SecretUpdate:
    """Update model for modifying secrets."""

class SecretFilter:
    """Filter for querying secrets."""

Service Connector Models

class ServiceConnectorRequest:
    """Request model for creating service connectors."""

class ServiceConnectorResponse:
    """Response model with connector information."""

class ServiceConnectorUpdate:
    """Update model for modifying connectors."""

class ServiceConnectorFilter:
    """Filter for querying connectors."""

Tag Models

class TagRequest:
    """Request model for creating tags."""

class TagResponse:
    """
    Response model with tag information.

    Attributes:
    - id: Tag UUID
    - name: Tag name
    - color: Tag color
    - tagged_count: Number of tagged resources
    """

class TagUpdate:
    """Update model for modifying tags."""

class TagFilter:
    """Filter for querying tags."""

Filter Models

Specialized filter classes for querying.

class StrFilter:
    """String filtering operations."""

class BoolFilter:
    """Boolean filtering operations."""

class NumericFilter:
    """Numeric filtering operations."""

class UUIDFilter:
    """UUID filtering operations."""

Service Connector Type Models

class ServiceConnectorTypeModel:
    """Service connector type definition."""

class AuthenticationMethodModel:
    """Authentication method definition."""

class ResourceTypeModel:
    """Resource type definition."""

class ServiceConnectorResourcesModel:
    """Available resources for connector."""

Usage Examples

Working with Response Models

from zenml.client import Client

client = Client()

# Get pipeline run (returns PipelineRunResponse)
run = client.get_pipeline_run("run_id")

# Access response attributes
print(f"Run ID: {run.id}")
print(f"Name: {run.name}")
print(f"Status: {run.status}")
print(f"Start time: {run.start_time}")

# Access nested models
print(f"Pipeline: {run.pipeline.name}")
print(f"Stack: {run.stack.name}")
print(f"User: {run.user.name}")

Using Filters

from zenml.client import Client

client = Client()

# Filter pipeline runs
runs = client.list_pipeline_runs(
    pipeline_id="pipeline_id",
    status="completed",
    sort_by="created",
    size=10
)

# Access paginated results
print(f"Total runs: {runs.total}")
print(f"Page: {runs.index}/{runs.total_pages}")

for run in runs.items:
    print(f"Run: {run.name}, Status: {run.status}")

Creating Resources

from zenml.client import Client

client = Client()

# Create stack (uses StackRequest internally)
stack = client.create_stack(
    name="my_stack",
    components={
        "orchestrator": "local",
        "artifact_store": "local"
    },
    description="Development stack"
)

# Returns StackResponse
print(f"Created stack: {stack.id}")
print(f"Components: {stack.components}")

Updating Resources

from zenml.client import Client

client = Client()

# Update model version (uses ModelVersionUpdate internally)
updated = client.update_model_version(
    model_name_or_id="my_model",
    version_name_or_id="1.0.0",
    stage="production",
    add_tags=["promoted", "validated"]
)

# Returns ModelVersionResponse
print(f"Updated version: {updated.version}")
print(f"Stage: {updated.stage}")
print(f"Tags: {updated.tags}")

Accessing Nested Models

from zenml.client import Client

client = Client()

# Get step run with nested artifact information
step_run = client.get_run_step("step_run_id")

# Access output artifacts (ArtifactVersionResponse models)
for output_name, artifact in step_run.outputs.items():
    print(f"Output: {output_name}")
    print(f"  Artifact: {artifact.artifact.name}")
    print(f"  Version: {artifact.version}")
    print(f"  Type: {artifact.type}")
    print(f"  URI: {artifact.uri}")

Model Serialization

from zenml.client import Client

client = Client()

# Get model (Pydantic model)
model = client.get_model("my_model")

# Serialize to dict
model_dict = model.dict()
print(model_dict)

# Serialize to JSON
model_json = model.json()
print(model_json)

# Model has Pydantic validation
try:
    invalid_model = model.copy(update={"invalid_field": "value"})
except Exception as e:
    print(f"Validation error: {e}")

All models in zenml.models are Pydantic models with automatic validation, serialization, and type checking. They represent the complete API surface for interacting with ZenML resources programmatically.