Quickstart

from typing import Optional
from zenml import step

@step
def simple_step(name: Optional[str] = None) -> str:
    """A simple step that returns a personalized greeting.
    
    This is the simplest possible ZenML step. It:
    1. Takes an optional input parameter
    2. Returns a string
    3. Is automatically tracked as an artifact by ZenML
    
    Args:
        name: Optional name to personalize the greeting
        
    Returns:
        A personalized greeting message
    """
    if name:
        message = f"Hello {name}! Welcome to ZenML 🚀"
    else:
        message = "Hello from ZenML! 🚀"
    print(message)
    return message

from typing import Annotated, Optional
from zenml import pipeline, step

@step
def simple_step(name: Optional[str] = None) -> str:
    """A simple step that returns a personalized greeting."""
    if name:
        message = f"Hello {name}! Welcome to ZenML 🚀"
    else:
        message = "Hello from ZenML! 🚀"
    print(message)
    return message

@pipeline(enable_cache=False)
def simple_pipeline(name: Optional[str] = None) -> Annotated[str, "greeting"]:
    """A simple pipeline that demonstrates ZenML basics.
    
    This pipeline:
    1. Takes an optional name parameter
    2. Calls a single step that returns a personalized greeting
    3. Returns the result as a tracked artifact
    
    Args:
        name: Optional name to personalize the greeting
        
    Returns:
        A greeting message as an artifact
    """
    greeting = simple_step(name=name)
    return greeting

if __name__ == "__main__":
    # Run the pipeline
    result = simple_pipeline()
    print(f"Pipeline returned: {result}")

from typing import Annotated, Dict, List
from zenml import pipeline, step
import random

@step
def generate_numbers(count: int = 10) -> Annotated[List[int], "random_numbers"]:
    """Generate a list of random numbers.
    
    Args:
        count: Number of random integers to generate
        
    Returns:
        List of random integers
    """
    numbers = [random.randint(1, 100) for _ in range(count)]
    print(f"Generated {count} numbers: {numbers}")
    return numbers

@step
def calculate_statistics(
    numbers: List[int]
) -> Annotated[Dict[str, float], "statistics"]:
    """Calculate statistics from a list of numbers.
    
    Args:
        numbers: List of integers to analyze
        
    Returns:
        Dictionary containing mean, median, min, and max
    """
    sorted_numbers = sorted(numbers)
    n = len(numbers)
    
    stats = {
        "mean": sum(numbers) / n,
        "median": sorted_numbers[n // 2],
        "min": min(numbers),
        "max": max(numbers),
        "count": n
    }
    
    print(f"Statistics: {stats}")
    return stats

@step
def format_report(
    statistics: Dict[str, float]
) -> Annotated[str, "report"]:
    """Create a formatted report from statistics.
    
    Args:
        statistics: Dictionary of calculated statistics
        
    Returns:
        Formatted report string
    """
    report = f"""
    📊 Statistics Report
    {'='*40}
    Count:  {statistics['count']}
    Mean:   {statistics['mean']:.2f}
    Median: {statistics['median']:.2f}
    Min:    {statistics['min']:.2f}
    Max:    {statistics['max']:.2f}
    """
    print(report)
    return report

@pipeline(enable_cache=True)
def statistics_pipeline(
    count: int = 10
) -> Annotated[str, "final_report"]:
    """Pipeline that generates numbers and computes statistics.
    
    This demonstrates:
    - Multiple steps with dependencies
    - Artifact passing between steps
    - Named artifacts with type annotations
    - Pipeline-level caching
    
    Args:
        count: Number of random integers to generate
        
    Returns:
        Formatted statistics report
    """
    numbers = generate_numbers(count=count)
    stats = calculate_statistics(numbers=numbers)
    report = format_report(statistics=stats)
    return report

if __name__ == "__main__":
    # Run with default parameters
    print("🚀 Running statistics pipeline...\n")
    result = statistics_pipeline()
    
    # Run again with different parameters
    print("\n🔄 Running again with 20 numbers...\n")
    result = statistics_pipeline(count=20)