Services

class BaseService:
    """
    Abstract base class for services.

    Services are long-running processes managed by ZenML,
    such as model deployment servers, feature store servers, etc.

    Attributes:
    - uuid: Service UUID
    - config: Service configuration
    - status: Service status object
    """

    def start(self):
        """Start the service."""

    def stop(self):
        """Stop the service."""

    def is_running(self) -> bool:
        """Check if service is running."""

    def get_logs(self) -> str:
        """Get service logs."""

from zenml.services import BaseService

class ServiceConfig:
    """
    Configuration for services.

    Base configuration class for all service types.

    Attributes:
    - name: Service name
    - description: Service description
    - pipeline_name: Associated pipeline name
    - pipeline_run_id: Associated pipeline run ID
    """

from zenml.services import ServiceConfig

class ServiceStatus:
    """
    Status tracking for services.

    Tracks the current state and health of a service.

    Attributes:
    - state: Service state (from ServiceState enum)
    - last_state: Previous state
    - error: Error message if in error state
    """

from zenml.services import ServiceStatus

class BaseServiceEndpoint:
    """
    Abstract base for service endpoints.

    Represents a network endpoint where a service is accessible.
    """

class ServiceEndpointConfig:
    """Configuration for service endpoints."""

class ServiceEndpointStatus:
    """Status tracking for service endpoints."""

class ServiceEndpointProtocol(str, Enum):
    """
    Protocol types.

    Values:
    - HTTP: HTTP protocol
    - HTTPS: HTTPS protocol
    - TCP: TCP protocol
    - UDP: UDP protocol
    - GRPC: gRPC protocol
    """
    HTTP = "http"
    HTTPS = "https"
    TCP = "tcp"
    UDP = "udp"
    GRPC = "grpc"

from zenml.services import (
    BaseServiceEndpoint,
    ServiceEndpointConfig,
    ServiceEndpointProtocol
)

class BaseServiceEndpointHealthMonitor:
    """Base for health monitoring."""

class HTTPEndpointHealthMonitor(BaseServiceEndpointHealthMonitor):
    """HTTP health monitor."""

class TCPEndpointHealthMonitor(BaseServiceEndpointHealthMonitor):
    """TCP health monitor."""

from zenml.services import (
    BaseServiceEndpointHealthMonitor,
    HTTPEndpointHealthMonitor,
    TCPEndpointHealthMonitor
)

class ContainerService(BaseService):
    """
    Service running in a container.

    Manages services running as Docker containers.
    """

class ContainerServiceConfig(ServiceConfig):
    """Configuration for container services."""

class ContainerServiceStatus(ServiceStatus):
    """Status for container services."""

from zenml.services import (
    ContainerService,
    ContainerServiceConfig
)

class LocalDaemonService(BaseService):
    """
    Service running as a local daemon.

    Manages services running as background processes on the local machine.
    """

class LocalDaemonServiceConfig(ServiceConfig):
    """Configuration for local daemon services."""

class LocalDaemonServiceStatus(ServiceStatus):
    """Status for local daemon services."""

from zenml.services import (
    LocalDaemonService,
    LocalDaemonServiceConfig
)

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"

from zenml.enums import ServiceState

from zenml.client import Client

client = Client()

# List all services
services = client.list_services()
for service in services:
    print(f"Service: {service.name}")
    print(f"Type: {service.service_type}")
    print(f"State: {service.state}")

# Get specific service
service = client.get_service("model_server_123")
print(f"Service: {service.name}")
print(f"Config: {service.config}")

# Delete service
client.delete_service("old_service_id")

from zenml.services import BaseService, ServiceState

# Get service instance
service = get_my_service()  # Implementation specific

# Check status
if service.status.state == ServiceState.INACTIVE:
    print("Service is not running")

# Start service
service.start()

# Wait for service to be active
while service.status.state == ServiceState.PENDING_STARTUP:
    time.sleep(1)

if service.status.state == ServiceState.ACTIVE:
    print("Service is running")

# Get logs
logs = service.get_logs()
print(logs)

# Stop service
service.stop()

from zenml.services import (
    BaseService,
    ServiceEndpointProtocol,
    HTTPEndpointHealthMonitor
)

# Service with HTTP endpoint
service = get_model_server()  # Implementation specific

# Access endpoint
endpoint = service.endpoint
print(f"Service URL: {endpoint.protocol}://{endpoint.hostname}:{endpoint.port}")

# Check health
health_monitor = HTTPEndpointHealthMonitor(
    endpoint=endpoint,
    healthcheck_uri="/health"
)

is_healthy = health_monitor.check_health()
print(f"Service healthy: {is_healthy}")

from zenml import step, pipeline
from zenml.client import Client

@step
def deploy_model(model: dict) -> str:
    """Deploy model and return service ID."""
    # Deploy model (integration-specific)
    # Returns service ID
    return "service_123"

@pipeline
def deployment_pipeline():
    model = {"weights": [0.1, 0.2]}
    service_id = deploy_model(model)
    return service_id

# Run pipeline
deployment_pipeline()

# Get deployed service
client = Client()
service = client.get_service("service_123")
print(f"Model service: {service.name}")

from zenml.client import Client
from zenml.enums import ServiceState

client = Client()

# Get all active services
services = client.list_services()
active_services = [
    s for s in services
    if s.state == ServiceState.ACTIVE
]

print(f"Active services: {len(active_services)}")

# Find services in error state
error_services = [
    s for s in services
    if s.state == ServiceState.ERROR
]

for service in error_services:
    print(f"Service {service.name} in error: {service.error}")

from zenml.client import Client
from zenml.enums import ServiceState

client = Client()

# Clean up inactive services
services = client.list_services()
for service in services:
    if service.state == ServiceState.INACTIVE:
        print(f"Deleting inactive service: {service.name}")
        client.delete_service(service.id)