Exceptions

class ZenMLBaseException(Exception):
    """
    Base exception for all ZenML exceptions.

    All custom ZenML exceptions inherit from this class,
    making it easy to catch any ZenML-specific error.

    Example:
    ```python
    try:
        # ZenML operations
        pass
    except ZenMLBaseException as e:
        print(f"ZenML error: {e}")
    ```
    """

from zenml.exceptions import ZenMLBaseException

class AuthorizationException(ZenMLBaseException):
    """
    Authorization/access errors.

    Raised when user lacks permissions for an operation.
    """

class CredentialsNotValid(AuthorizationException):
    """
    Invalid credentials error.

    Raised when credentials are invalid or expired.
    Triggers re-authentication flow.
    """

class OAuthError(ZenMLBaseException):
    """
    OAuth2 authentication errors.

    Raised during OAuth2 authentication flows.
    """

from zenml.exceptions import (
    AuthorizationException,
    CredentialsNotValid,
    OAuthError
)

class DoesNotExistException(ZenMLBaseException):
    """
    Entity not found errors.

    Raised when attempting to access a non-existent resource.
    """

class EntityExistsError(ZenMLBaseException):
    """
    Entity already exists errors.

    Raised when attempting to create a resource that already exists.
    """

class EntityCreationError(ZenMLBaseException):
    """
    Entity creation failures.

    Raised when entity creation fails due to validation or constraints.
    """

from zenml.exceptions import (
    DoesNotExistException,
    EntityExistsError,
    EntityCreationError
)

class StepInterfaceError(ZenMLBaseException):
    """
    Step interface usage errors.

    Raised when step decorators or interfaces are used incorrectly.
    """

class StepContextError(ZenMLBaseException):
    """
    Step context access errors.

    Raised when attempting to access step context outside execution.
    """

class InputResolutionError(ZenMLBaseException):
    """
    Step input resolution failures.

    Raised when step inputs cannot be resolved or connected.
    """

class RunStoppedException(ZenMLBaseException):
    """
    Pipeline run stopped by user.

    Raised when a pipeline run is explicitly stopped.
    """

class RunInterruptedException(ZenMLBaseException):
    """
    Step run interrupted.

    Raised when step execution is interrupted.
    """

from zenml.exceptions import (
    StepInterfaceError,
    StepContextError,
    InputResolutionError,
    RunStoppedException,
    RunInterruptedException
)

class StackComponentInterfaceError(ZenMLBaseException):
    """
    Stack component interface errors.

    Raised when stack component interfaces are used incorrectly.
    """

class StackValidationError(ZenMLBaseException):
    """
    Stack configuration validation errors.

    Raised when stack configuration is invalid or incompatible.
    """

class ArtifactStoreInterfaceError(ZenMLBaseException):
    """
    Artifact store interface errors.

    Raised when artifact store operations fail.
    """

from zenml.exceptions import (
    StackComponentInterfaceError,
    StackValidationError,
    ArtifactStoreInterfaceError
)

class MaterializerInterfaceError(ZenMLBaseException):
    """
    Materializer interface errors.

    Raised when materializer operations fail or are used incorrectly.
    """

from zenml.exceptions import MaterializerInterfaceError

class IntegrationError(ZenMLBaseException):
    """
    Integration activation errors.

    Raised when integration requirements are not met or activation fails.
    """

class CustomFlavorImportError(ZenMLBaseException):
    """
    Custom flavor import failures.

    Raised when custom flavor cannot be imported.
    """

from zenml.exceptions import IntegrationError, CustomFlavorImportError

class ValidationError(ZenMLBaseException):
    """
    Model/data validation errors.

    Raised when Pydantic model validation fails or data is invalid.
    """

class HydrationError(ZenMLBaseException):
    """
    Model hydration failures.

    Raised when model objects cannot be hydrated from stored data.
    """

class HookValidationException(ZenMLBaseException):
    """
    Hook validation failures.

    Raised when hook specifications are invalid.
    """

from zenml.exceptions import (
    ValidationError,
    HydrationError,
    HookValidationException
)

class SettingsResolvingError(ZenMLBaseException):
    """
    Settings resolution failures.

    Raised when step or pipeline settings cannot be resolved.
    """

class SecretsStoreNotConfiguredError(ZenMLBaseException):
    """
    Secrets store not configured.

    Raised when secrets operations attempted without configured store.
    """

class BackupSecretsStoreNotConfiguredError(ZenMLBaseException):
    """
    Backup secrets store not configured.

    Raised when backup secrets operations attempted without configured store.
    """

from zenml.exceptions import (
    SettingsResolvingError,
    SecretsStoreNotConfiguredError,
    BackupSecretsStoreNotConfiguredError
)

class IllegalOperationError(ZenMLBaseException):
    """
    Illegal operation attempts.

    Raised when attempting operations that are not allowed in current context.
    """

class MethodNotAllowedError(ZenMLBaseException):
    """
    HTTP method not allowed.

    Raised for REST API method errors.
    """

class MaxConcurrentTasksError(ZenMLBaseException):
    """
    Concurrent task limit reached.

    Raised when maximum concurrent tasks limit is exceeded.
    """

class WebhookInactiveError(ZenMLBaseException):
    """
    Inactive webhook errors.

    Raised when attempting to use inactive webhook.
    """

from zenml.exceptions import (
    IllegalOperationError,
    MethodNotAllowedError,
    MaxConcurrentTasksError,
    WebhookInactiveError
)

class InitializationException(ZenMLBaseException):
    """
    Repository initialization errors.

    Raised when ZenML repository initialization fails.
    """

class GitNotFoundError(ZenMLBaseException):
    """
    Git not installed errors.

    Raised when Git is required but not found on system.
    """

class SubscriptionUpgradeRequiredError(ZenMLBaseException):
    """
    Subscription tier limitation.

    Raised when feature requires higher subscription tier.
    """

class RunMonitoringError(ZenMLBaseException):
    """
    Pipeline run monitoring errors.

    Raised when run monitoring operations fail.
    """

class HookExecutionException(ZenMLBaseException):
    """
    Hook execution failures.

    Raised when hook execution fails during pipeline/step execution.
    """

from zenml.exceptions import (
    InitializationException,
    GitNotFoundError,
    SubscriptionUpgradeRequiredError,
    RunMonitoringError,
    HookExecutionException
)

class ZenKeyError(KeyError, ZenMLBaseException):
    """
    KeyError with multiline message support.

    Enhanced KeyError that supports detailed error messages.
    """

from zenml.exceptions import ZenKeyError

from zenml.client import Client
from zenml.exceptions import ZenMLBaseException, DoesNotExistException

client = Client()

try:
    pipeline = client.get_pipeline("nonexistent_pipeline")
except DoesNotExistException as e:
    print(f"Pipeline not found: {e}")
except ZenMLBaseException as e:
    print(f"ZenML error: {e}")

from zenml.client import Client
from zenml.exceptions import AuthorizationException, CredentialsNotValid

try:
    client = Client()
    # Perform authenticated operation
    client.list_pipelines()
except CredentialsNotValid:
    print("Credentials expired. Please re-authenticate.")
    # Trigger re-authentication
except AuthorizationException as e:
    print(f"Not authorized: {e}")

from zenml.client import Client
from zenml.exceptions import EntityExistsError, EntityCreationError

client = Client()

try:
    stack = client.create_stack(
        name="my_stack",
        components={"orchestrator": "local", "artifact_store": "local"}
    )
except EntityExistsError:
    print("Stack already exists")
    stack = client.get_stack("my_stack")
except EntityCreationError as e:
    print(f"Failed to create stack: {e}")

from zenml import get_step_context
from zenml.exceptions import StepContextError

try:
    context = get_step_context()
    print(context.step_name)
except StepContextError:
    print("Not inside step execution")

from zenml.client import Client
from zenml.exceptions import StackValidationError

client = Client()

try:
    client.create_stack(
        name="invalid_stack",
        components={}  # Missing required components
    )
except StackValidationError as e:
    print(f"Stack validation failed: {e}")

from zenml.integrations.aws import AWSIntegration
from zenml.exceptions import IntegrationError

try:
    if not AWSIntegration.check_installation():
        raise IntegrationError("AWS integration not installed")
except IntegrationError as e:
    print(f"Integration error: {e}")
    print("Install with: pip install zenml[aws]")

from zenml import pipeline
from zenml.exceptions import HookExecutionException

def my_hook():
    raise ValueError("Hook failed")

@pipeline(on_success=my_hook)
def my_pipeline():
    pass

try:
    my_pipeline()
except HookExecutionException as e:
    print(f"Hook execution failed: {e}")

from zenml import pipeline, step
from zenml.exceptions import (
    ZenMLBaseException,
    StepInterfaceError,
    InputResolutionError,
    RunStoppedException,
    StackValidationError
)

@step
def risky_step(data: list) -> dict:
    if not data:
        raise ValueError("Empty data")
    return {"processed": data}

@pipeline
def error_handling_pipeline():
    try:
        data = []
        result = risky_step(data)
    except StepInterfaceError as e:
        print(f"Step interface error: {e}")
    except InputResolutionError as e:
        print(f"Input resolution error: {e}")
    except ZenMLBaseException as e:
        print(f"ZenML error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")

from zenml.exceptions import ZenMLBaseException

class CustomPipelineError(ZenMLBaseException):
    """Custom exception for pipeline-specific errors."""
    pass

def validate_data(data):
    if not isinstance(data, list):
        raise CustomPipelineError(
            f"Expected list, got {type(data).__name__}"
        )

try:
    validate_data("not a list")
except CustomPipelineError as e:
    print(f"Validation error: {e}")