Deployers

zenml.deployers

Deployers are stack components responsible for deploying pipelines as HTTP services.

Deploying pipelines is the process of hosting and running machine-learning pipelines as part of a managed web service and providing access to pipeline execution through an API endpoint like HTTP or GRPC. Once deployed, you can send execution requests to the pipeline through the web service's API and receive responses containing the pipeline results or execution status.

Add a deployer to your ZenML stack to be able to provision pipelines deployments that transform your ML pipelines into long-running HTTP services for real-time, on-demand execution instead of traditional batch processing.

When present in a stack, the deployer also acts as a registry for pipeline endpoints that are deployed with ZenML. You can use the deployer to list all deployments that are currently provisioned for online execution or filtered according to a particular snapshot or configuration, or to delete an external deployment managed through ZenML.

Attributes

__all__ = ['BaseDeployer', 'BaseDeployerFlavor', 'BaseDeployerConfig', 'ContainerizedDeployer', 'DockerDeployer', 'DockerDeployerFlavor'] module-attribute

Classes

BaseDeployer(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Bases: StackComponent, ABC

Base class for all ZenML deployers.

The deployer serves three major purposes:

1. It contains all the stack related configuration attributes required to interact with the remote pipeline deployment tool, service or platform (e.g. hostnames, URLs, references to credentials, other client related configuration parameters).

2. It implements the life-cycle management for deployments, including discovery, creation, deletion and updating.

3. It acts as a ZenML deployment registry, where every pipeline deployment is stored as a database entity through the ZenML Client. This allows the deployer to keep track of all externally running pipeline deployments and to manage their lifecycle.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        environment: Environment variables to set when running on this
            component.
        secrets: Secrets to set as environment variables when running on
            this component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.environment = environment or {}
    self.secrets = secrets or []
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None

Attributes

config: BaseDeployerConfig property

Returns the BaseDeployerConfig config.

Returns:

Type	Description
BaseDeployerConfig	The configuration.

Functions

delete_deployment(deployment_name_or_id: Union[str, UUID], project: Optional[UUID] = None, force: bool = False, timeout: Optional[int] = None) -> None

Deprovision and delete a deployment.

Parameters:

Name	Type	Description	Default
deployment_name_or_id	Union[str, UUID]	The name or ID of the deployment to delete.	required
project	Optional[UUID]	The project ID of the deployment to deprovision. Required if a name is provided.	None
force	bool	if True, force the deployment to delete even if it cannot be deprovisioned.	False
timeout	Optional[int]	The maximum time in seconds to wait for the pipeline deployment to be deprovisioned. If provided, will override the deployer's default timeout.	None

Raises:

Type	Description
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

def delete_deployment(
    self,
    deployment_name_or_id: Union[str, UUID],
    project: Optional[UUID] = None,
    force: bool = False,
    timeout: Optional[int] = None,
) -> None:
    """Deprovision and delete a deployment.

    Args:
        deployment_name_or_id: The name or ID of the deployment to
            delete.
        project: The project ID of the deployment to deprovision.
            Required if a name is provided.
        force: if True, force the deployment to delete even if it
            cannot be deprovisioned.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be deprovisioned. If provided, will override the
            deployer's default timeout.

    Raises:
        DeployerError: if an unexpected error occurs.
    """
    client = Client()
    try:
        deployment = self.deprovision_deployment(
            deployment_name_or_id, project, timeout
        )
    except DeploymentNotFoundError:
        # The deployment was already deleted
        return
    except DeployerError as e:
        if force:
            logger.warning(
                f"Failed to deprovision deployment "
                f"{deployment_name_or_id}: {e}. Forcing deletion."
            )
            deployment = client.get_deployment(
                deployment_name_or_id, project=project
            )
            client.zen_store.delete_deployment(deployment_id=deployment.id)
        else:
            raise
    else:
        client.zen_store.delete_deployment(deployment_id=deployment.id)

deprovision_deployment(deployment_name_or_id: Union[str, UUID], project: Optional[UUID] = None, timeout: Optional[int] = None) -> DeploymentResponse

Deprovision a deployment.

Parameters:

Name	Type	Description	Default
deployment_name_or_id	Union[str, UUID]	The name or ID of the deployment to deprovision.	required
project	Optional[UUID]	The project ID of the deployment to deprovision. Required if a name is provided.	None
timeout	Optional[int]	The maximum time in seconds to wait for the pipeline deployment to deprovision. If provided, will override the deployer's default timeout.	None

Returns:

Type	Description
DeploymentResponse	The deployment.

Raises:

Type	Description
DeploymentNotFoundError	if the deployment is not found or is not managed by this deployer.
DeploymentDeprovisionError	if the deployment deprovision fails.
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

def deprovision_deployment(
    self,
    deployment_name_or_id: Union[str, UUID],
    project: Optional[UUID] = None,
    timeout: Optional[int] = None,
) -> DeploymentResponse:
    """Deprovision a deployment.

    Args:
        deployment_name_or_id: The name or ID of the deployment to
            deprovision.
        project: The project ID of the deployment to deprovision.
            Required if a name is provided.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to deprovision. If provided, will override the
            deployer's default timeout.

    Returns:
        The deployment.

    Raises:
        DeploymentNotFoundError: if the deployment is not found
            or is not managed by this deployer.
        DeploymentDeprovisionError: if the deployment
            deprovision fails.
        DeployerError: if an unexpected error occurs.
    """
    client = Client()
    try:
        deployment = client.get_deployment(
            deployment_name_or_id, project=project
        )
    except KeyError:
        raise DeploymentNotFoundError(
            f"Deployment with name or ID '{deployment_name_or_id}' "
            f"not found"
        )

    self._check_deployment_deployer(deployment)

    if not timeout and deployment.snapshot:
        settings = cast(
            BaseDeployerSettings,
            self.get_settings(deployment.snapshot),
        )

        timeout = settings.lcm_timeout

    timeout = timeout or DEFAULT_DEPLOYMENT_LCM_TIMEOUT

    start_time = time.time()
    deployment_state = DeploymentOperationalState(
        status=DeploymentStatus.ERROR,
    )
    with track_handler(
        AnalyticsEvent.STOP_DEPLOYMENT
    ) as analytics_handler:
        try:
            deleted_deployment_state = self.do_deprovision_deployment(
                deployment, timeout
            )
            if not deleted_deployment_state:
                # When do_delete_deployment returns a None value, this
                # is to signal that the deployment is already fully deprovisioned.
                deployment_state.status = DeploymentStatus.ABSENT
        except DeploymentNotFoundError:
            deployment_state.status = DeploymentStatus.ABSENT
        except DeployerError as e:
            raise DeployerError(
                f"Failed to delete deployment {deployment_name_or_id}: {e}"
            ) from e
        except Exception as e:
            raise DeployerError(
                f"Unexpected error while deleting deployment for "
                f"{deployment_name_or_id}: {e}"
            ) from e
        finally:
            deployment = self._update_deployment(
                deployment, deployment_state
            )

        try:
            if deployment_state.status == DeploymentStatus.ABSENT:
                return deployment

            # Subtract the time spent deprovisioning the deployment from the timeout
            timeout = timeout - int(time.time() - start_time)
            deployment, _ = self._poll_deployment(
                deployment, DeploymentStatus.ABSENT, timeout
            )

            if deployment.status != DeploymentStatus.ABSENT:
                raise DeploymentDeprovisionError(
                    f"Failed to deprovision deployment {deployment_name_or_id}: "
                    f"Operational state: {deployment.status}"
                )

        finally:
            analytics_handler.metadata = (
                self._get_deployment_analytics_metadata(
                    deployment=deployment,
                    stack=None,
                )
            )

        return deployment

do_deprovision_deployment(deployment: DeploymentResponse, timeout: int) -> Optional[DeploymentOperationalState] abstractmethod

Abstract method to deprovision a deployment.

Concrete deployer subclasses must implement the following functionality in this method:

• Deprovision the actual deployment infrastructure (e.g., FastAPI server, Kubernetes deployment, cloud function, etc.) based on the information in the deployment response.

• Return a DeploymentOperationalState representing the operational state of the deleted deployment, or None if the deletion is completed before the call returns.

Note that the deployment infrastructure is not required to be deleted immediately. The deployer can return a DeploymentOperationalState with a status of DeploymentStatus.PENDING, and the base deployer will poll the deployment infrastructure by calling the do_get_deployment_state method until it is deleted or it times out.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to delete.	required
timeout	int	The maximum time in seconds to wait for the pipeline deployment to be deprovisioned.	required

Returns:

Type	Description
Optional[DeploymentOperationalState]	The DeploymentOperationalState object representing the
Optional[DeploymentOperationalState]	operational state of the deprovisioned deployment, or None
Optional[DeploymentOperationalState]	if the deprovision is completed before the call returns.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.
DeploymentDeprovisionError	if the deployment deprovision fails.
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

@abstractmethod
def do_deprovision_deployment(
    self,
    deployment: DeploymentResponse,
    timeout: int,
) -> Optional[DeploymentOperationalState]:
    """Abstract method to deprovision a deployment.

    Concrete deployer subclasses must implement the following
    functionality in this method:

    - Deprovision the actual deployment infrastructure (e.g.,
    FastAPI server, Kubernetes deployment, cloud function, etc.) based on
    the information in the deployment response.

    - Return a DeploymentOperationalState representing the operational
    state of the deleted deployment, or None if the deletion is
    completed before the call returns.

    Note that the deployment infrastructure is not required to be
    deleted immediately. The deployer can return a
    DeploymentOperationalState with a status of
    DeploymentStatus.PENDING, and the base deployer will poll
    the deployment infrastructure by calling the
    `do_get_deployment_state` method until it is deleted or it times out.

    Args:
        deployment: The deployment to delete.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be deprovisioned.

    Returns:
        The DeploymentOperationalState object representing the
        operational state of the deprovisioned deployment, or None
        if the deprovision is completed before the call returns.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
        DeploymentDeprovisionError: if the deployment
            deprovision fails.
        DeployerError: if an unexpected error occurs.
    """

do_get_deployment_state(deployment: DeploymentResponse) -> DeploymentOperationalState abstractmethod

Abstract method to get information about a deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to get information about.	required

Returns:

Type	Description
DeploymentOperationalState	The DeploymentOperationalState object representing the
DeploymentOperationalState	updated operational state of the deployment.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.
DeployerError	if the deployment information cannot be retrieved for any other reason or if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

@abstractmethod
def do_get_deployment_state(
    self,
    deployment: DeploymentResponse,
) -> DeploymentOperationalState:
    """Abstract method to get information about a deployment.

    Args:
        deployment: The deployment to get information about.

    Returns:
        The DeploymentOperationalState object representing the
        updated operational state of the deployment.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
        DeployerError: if the deployment information cannot
            be retrieved for any other reason or if an unexpected error
            occurs.
    """

do_get_deployment_state_logs(deployment: DeploymentResponse, follow: bool = False, tail: Optional[int] = None) -> Generator[str, bool, None] abstractmethod

Abstract method to get the logs of a deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to get the logs of.	required
follow	bool	if True, the logs will be streamed as they are written	False
tail	Optional[int]	only retrieve the last NUM lines of log output.	None

Yields:

Type	Description
str	The logs of the deployment.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.
DeploymentLogsNotFoundError	if the deployment logs are not found.
DeployerError	if the deployment logs cannot be retrieved for any other reason or if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

@abstractmethod
def do_get_deployment_state_logs(
    self,
    deployment: DeploymentResponse,
    follow: bool = False,
    tail: Optional[int] = None,
) -> Generator[str, bool, None]:
    """Abstract method to get the logs of a deployment.

    Args:
        deployment: The deployment to get the logs of.
        follow: if True, the logs will be streamed as they are written
        tail: only retrieve the last NUM lines of log output.

    Yields:
        The logs of the deployment.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
        DeploymentLogsNotFoundError: if the deployment logs are not
            found.
        DeployerError: if the deployment logs cannot
            be retrieved for any other reason or if an unexpected error
            occurs.
    """

do_provision_deployment(deployment: DeploymentResponse, stack: Stack, environment: Dict[str, str], secrets: Dict[str, str], timeout: int) -> DeploymentOperationalState abstractmethod

Abstract method to deploy a pipeline as an HTTP deployment.

Concrete deployer subclasses must implement the following functionality in this method:

• Create the actual deployment infrastructure (e.g., FastAPI server, Kubernetes deployment, cloud function, etc.) based on the information in the deployment response, particularly the pipeline snapshot. When determining how to name the external resources, do not rely on the deployment name as being immutable or unique.

• If the deployment infrastructure is already provisioned, update it to match the information in the deployment response.

• Return a DeploymentOperationalState representing the operational state of the provisioned deployment.

Note that the deployment infrastructure is not required to be deployed immediately. The deployer can return a DeploymentOperationalState with a status of DeploymentStatus.PENDING, and the base deployer will poll the deployment infrastructure by calling the do_get_deployment_state method until it is ready or it times out.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to deploy as an HTTP deployment.	required
stack	Stack	The stack the pipeline will be deployed on.	required
environment	Dict[str, str]	A dictionary of environment variables to set on the deployment.	required
secrets	Dict[str, str]	A dictionary of secret environment variables to set on the deployment. These secret environment variables should not be exposed as regular environment variables on the deployer.	required
timeout	int	The maximum time in seconds to wait for the pipeline deployment to be provisioned.	required

Returns:

Type	Description
DeploymentOperationalState	The DeploymentOperationalState object representing the
DeploymentOperationalState	operational state of the provisioned deployment.

Raises:

Type	Description
DeploymentProvisionError	if provisioning the deployment fails.
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

@abstractmethod
def do_provision_deployment(
    self,
    deployment: DeploymentResponse,
    stack: "Stack",
    environment: Dict[str, str],
    secrets: Dict[str, str],
    timeout: int,
) -> DeploymentOperationalState:
    """Abstract method to deploy a pipeline as an HTTP deployment.

    Concrete deployer subclasses must implement the following
    functionality in this method:

    - Create the actual deployment infrastructure (e.g.,
    FastAPI server, Kubernetes deployment, cloud function, etc.) based on
    the information in the deployment response, particularly the
    pipeline snapshot. When determining how to name the external
    resources, do not rely on the deployment name as being immutable
    or unique.

    - If the deployment infrastructure is already provisioned, update
    it to match the information in the deployment response.

    - Return a DeploymentOperationalState representing the operational
    state of the provisioned deployment.

    Note that the deployment infrastructure is not required to be
    deployed immediately. The deployer can return a
    DeploymentOperationalState with a status of
    DeploymentStatus.PENDING, and the base deployer will poll
    the deployment infrastructure by calling the
    `do_get_deployment_state` method until it is ready or it times out.

    Args:
        deployment: The deployment to deploy as an HTTP deployment.
        stack: The stack the pipeline will be deployed on.
        environment: A dictionary of environment variables to set on the
            deployment.
        secrets: A dictionary of secret environment variables to set
            on the deployment. These secret environment variables
            should not be exposed as regular environment variables on the
            deployer.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be provisioned.

    Returns:
        The DeploymentOperationalState object representing the
        operational state of the provisioned deployment.

    Raises:
        DeploymentProvisionError: if provisioning the deployment
            fails.
        DeployerError: if an unexpected error occurs.
    """

get_active_deployer() -> BaseDeployer classmethod

Get the deployer registered in the active stack.

Returns:

Type	Description
BaseDeployer	The deployer registered in the active stack.

Raises:

Type	Description
TypeError	if a deployer is not part of the active stack.

Source code in src/zenml/deployers/base_deployer.py

@classmethod
def get_active_deployer(cls) -> "BaseDeployer":
    """Get the deployer registered in the active stack.

    Returns:
        The deployer registered in the active stack.

    Raises:
        TypeError: if a deployer is not part of the
            active stack.
    """
    client = Client()
    deployer = client.active_stack.deployer
    if not deployer or not isinstance(deployer, cls):
        raise TypeError(
            "The active stack needs to have a deployer "
            "component registered to be able to deploy pipelines. "
            "You can create a new stack with a deployer component "
            "or update your active stack to add this component, e.g.:\n\n"
            "  `zenml deployer register ...`\n"
            "  `zenml stack register <STACK-NAME> -D ...`\n"
            "  or:\n"
            "  `zenml stack update -D ...`\n\n"
        )

    return deployer

get_deployment_logs(deployment_name_or_id: Union[str, UUID], project: Optional[UUID] = None, follow: bool = False, tail: Optional[int] = None) -> Generator[str, bool, None]

Get the logs of a deployment.

Parameters:

Name	Type	Description	Default
deployment_name_or_id	Union[str, UUID]	The name or ID of the deployment to get the logs of.	required
project	Optional[UUID]	The project ID of the deployment to get the logs of. Required if a name is provided.	None
follow	bool	if True, the logs will be streamed as they are written.	False
tail	Optional[int]	only retrieve the last NUM lines of log output.	None

Returns:

Type	Description
None	A generator that yields the logs of the deployment.

Raises:

Type	Description
DeploymentNotFoundError	if the deployment is not found.
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

def get_deployment_logs(
    self,
    deployment_name_or_id: Union[str, UUID],
    project: Optional[UUID] = None,
    follow: bool = False,
    tail: Optional[int] = None,
) -> Generator[str, bool, None]:
    """Get the logs of a deployment.

    Args:
        deployment_name_or_id: The name or ID of the deployment to get
            the logs of.
        project: The project ID of the deployment to get the logs of.
            Required if a name is provided.
        follow: if True, the logs will be streamed as they are written.
        tail: only retrieve the last NUM lines of log output.

    Returns:
        A generator that yields the logs of the deployment.

    Raises:
        DeploymentNotFoundError: if the deployment is not found.
        DeployerError: if an unexpected error occurs.
    """
    client = Client()
    try:
        deployment = client.get_deployment(
            deployment_name_or_id, project=project
        )
    except KeyError:
        raise DeploymentNotFoundError(
            f"Deployment with name or ID '{deployment_name_or_id}' "
            f"not found"
        )

    self._check_deployment_deployer(deployment)

    try:
        return self.do_get_deployment_state_logs(deployment, follow, tail)
    except DeployerError as e:
        raise DeployerError(
            f"Failed to get logs for deployment {deployment_name_or_id}: {e}"
        ) from e
    except Exception as e:
        raise DeployerError(
            f"Unexpected error while getting logs for deployment for "
            f"{deployment_name_or_id}: {e}"
        ) from e

provision_deployment(snapshot: PipelineSnapshotResponse, stack: Stack, deployment_name_or_id: Union[str, UUID], replace: bool = True, timeout: Optional[int] = None) -> DeploymentResponse

Provision a deployment.

The provision_deployment method is the main entry point for provisioning deployments using the deployer. It is used to deploy a pipeline snapshot as an HTTP deployment, or update an existing deployment instance with the same name. The method returns a DeploymentResponse object that is a representation of the external deployment instance.

Parameters:

Name	Type	Description	Default
snapshot	PipelineSnapshotResponse	The pipeline snapshot to deploy as an HTTP deployment.	required
stack	Stack	The stack the pipeline will be deployed on.	required
deployment_name_or_id	Union[str, UUID]	Unique name or ID for the deployment. This name must be unique at the project level.	required
replace	bool	If True, it will update in-place any existing pipeline deployment instance with the same name. If False, and the pipeline deployment instance already exists, it will raise a DeploymentAlreadyExistsError.	True
timeout	Optional[int]	The maximum time in seconds to wait for the pipeline deployment to be provisioned. If provided, will override the deployer's default timeout.	None

Raises:

Type	Description
DeploymentAlreadyExistsError	if the deployment already exists and replace is False.
DeploymentProvisionError	if the deployment fails.
DeploymentSnapshotMismatchError	if the pipeline snapshot was not created for this deployer.
DeploymentNotFoundError	if the deployment with the given ID is not found.
DeployerError	if an unexpected error occurs.

Returns:

Type	Description
DeploymentResponse	The DeploymentResponse object representing the provisioned
DeploymentResponse	deployment.

Source code in src/zenml/deployers/base_deployer.py

def provision_deployment(
    self,
    snapshot: PipelineSnapshotResponse,
    stack: "Stack",
    deployment_name_or_id: Union[str, UUID],
    replace: bool = True,
    timeout: Optional[int] = None,
) -> DeploymentResponse:
    """Provision a deployment.

    The provision_deployment method is the main entry point for
    provisioning deployments using the deployer. It is used to deploy
    a pipeline snapshot as an HTTP deployment, or update an existing
    deployment instance with the same name. The method returns a
    DeploymentResponse object that is a representation of the
    external deployment instance.

    Args:
        snapshot: The pipeline snapshot to deploy as an HTTP deployment.
        stack: The stack the pipeline will be deployed on.
        deployment_name_or_id: Unique name or ID for the deployment.
            This name must be unique at the project level.
        replace: If True, it will update in-place any existing pipeline
            deployment instance with the same name. If False, and the pipeline
            deployment instance already exists, it will raise a
            DeploymentAlreadyExistsError.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be provisioned. If provided, will override the
            deployer's default timeout.

    Raises:
        DeploymentAlreadyExistsError: if the deployment already
            exists and replace is False.
        DeploymentProvisionError: if the deployment fails.
        DeploymentSnapshotMismatchError: if the pipeline snapshot
            was not created for this deployer.
        DeploymentNotFoundError: if the deployment with the
            given ID is not found.
        DeployerError: if an unexpected error occurs.

    Returns:
        The DeploymentResponse object representing the provisioned
        deployment.
    """
    if not replace and is_valid_uuid(deployment_name_or_id):
        raise DeploymentAlreadyExistsError(
            f"A deployment with ID '{deployment_name_or_id}' "
            "already exists"
        )

    self._check_deployment_inputs_outputs(snapshot)

    client = Client()

    settings = cast(
        BaseDeployerSettings,
        self.get_settings(snapshot),
    )

    timeout = timeout or settings.lcm_timeout
    auth_key = settings.auth_key
    if not auth_key and settings.generate_auth_key:
        auth_key = self._generate_auth_key()

    if snapshot.stack and snapshot.stack.id != stack.id:
        # When a different stack is used then the one the snapshot was
        # created for, the container image may not have the correct
        # dependencies installed, which leads to unexpected errors during
        # deployment. To avoid this, we raise an error here.
        raise DeploymentSnapshotMismatchError(
            f"The pipeline snapshot with ID '{snapshot.id}' "
            f"was not created for the stack {stack.name} and might not "
            "have the correct dependencies installed. This may "
            "lead to unexpected behavior during deployment. Please switch "
            f"to the correct active stack '{snapshot.stack.name}' or use "
            "a different snapshot."
        )

    try:
        # Get the existing deployment
        deployment = client.get_deployment(
            deployment_name_or_id, project=snapshot.project_id
        )

        self._check_snapshot_already_deployed(snapshot, deployment.id)

        logger.debug(
            f"Existing deployment found with name '{deployment.name}'"
        )
    except KeyError:
        if isinstance(deployment_name_or_id, UUID):
            raise DeploymentNotFoundError(
                f"Deployment with ID '{deployment_name_or_id}' not found"
            )

        self._check_snapshot_already_deployed(
            snapshot, deployment_name_or_id
        )

        logger.debug(
            f"Creating new deployment {deployment_name_or_id} with "
            f"snapshot ID: {snapshot.id}"
        )

        # Create the deployment request
        deployment_request = DeploymentRequest(
            name=deployment_name_or_id,
            project=snapshot.project_id,
            snapshot_id=snapshot.id,
            deployer_id=self.id,  # This deployer's ID
            auth_key=auth_key,
        )

        deployment = client.zen_store.create_deployment(deployment_request)
        logger.debug(
            f"Created new deployment with name '{deployment.name}' "
            f"and ID: {deployment.id}"
        )
    else:
        if not replace:
            raise DeploymentAlreadyExistsError(
                f"A deployment with name '{deployment.name}' "
                "already exists"
            )

        self._check_deployment_deployer(deployment)
        self._check_deployment_snapshot(snapshot)

        deployment_update = DeploymentUpdate(
            snapshot_id=snapshot.id,
        )
        if (
            deployment.auth_key
            and not auth_key
            or not deployment.auth_key
            and auth_key
        ):
            # Key was either added or removed
            deployment_update.auth_key = auth_key
        elif deployment.auth_key != auth_key and (
            settings.auth_key or not settings.generate_auth_key
        ):
            # Key was changed and not because of re-generation
            deployment_update.auth_key = auth_key

        # The deployment has been updated
        deployment = client.zen_store.update_deployment(
            deployment.id,
            deployment_update,
        )

    logger.info(
        f"Provisioning deployment {deployment.name} with "
        f"snapshot ID: {snapshot.id}"
    )

    environment, secrets = get_config_environment_vars(
        deployment_id=deployment.id,
    )

    # Make sure to use the correct active stack/project which correspond
    # to the supplied stack and snapshot, which may be different from the
    # active stack/project
    environment[ENV_ZENML_ACTIVE_STACK_ID] = str(stack.id)
    environment[ENV_ZENML_ACTIVE_PROJECT_ID] = str(snapshot.project_id)

    start_time = time.time()
    deployment_state = DeploymentOperationalState(
        status=DeploymentStatus.ERROR,
    )
    with track_handler(
        AnalyticsEvent.DEPLOY_PIPELINE
    ) as analytics_handler:
        try:
            deployment_state = self.do_provision_deployment(
                deployment,
                stack=stack,
                environment=environment,
                secrets=secrets,
                timeout=timeout,
            )
        except DeploymentProvisionError as e:
            raise DeploymentProvisionError(
                f"Failed to provision deployment {deployment.name}: {e}"
            ) from e
        except DeployerError as e:
            raise DeployerError(
                f"Failed to provision deployment {deployment.name}: {e}"
            ) from e
        except Exception as e:
            raise DeployerError(
                f"Unexpected error while provisioning deployment for "
                f"{deployment.name}: {e}"
            ) from e
        finally:
            deployment = self._update_deployment(
                deployment, deployment_state
            )

        logger.info(
            f"Provisioned deployment {deployment.name} with "
            f"snapshot ID: {snapshot.id}. Operational state is: "
            f"{deployment_state.status}"
        )

        try:
            if deployment_state.status == DeploymentStatus.RUNNING:
                return deployment

            # Subtract the time spent deploying the deployment from the
            # timeout
            timeout = timeout - int(time.time() - start_time)
            deployment, _ = self._poll_deployment(
                deployment, DeploymentStatus.RUNNING, timeout
            )

            if deployment.status != DeploymentStatus.RUNNING:
                raise DeploymentProvisionError(
                    f"Failed to provision deployment {deployment.name}: "
                    f"The deployment's operational state is "
                    f"{deployment.status}. Please check the status or logs "
                    "of the deployment for more information."
                )

        finally:
            analytics_handler.metadata = (
                self._get_deployment_analytics_metadata(
                    deployment=deployment,
                    stack=stack,
                )
            )

        return deployment

refresh_deployment(deployment_name_or_id: Union[str, UUID], project: Optional[UUID] = None) -> DeploymentResponse

Refresh the status of a deployment by name or ID.

Call this to refresh the operational state of a deployment.

Parameters:

Name	Type	Description	Default
deployment_name_or_id	Union[str, UUID]	The name or ID of the deployment to get.	required
project	Optional[UUID]	The project ID of the deployment to get. Required if a name is provided.	None

Returns:

Type	Description
DeploymentResponse	The deployment.

Raises:

Type	Description
DeploymentNotFoundError	if the deployment is not found.
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

def refresh_deployment(
    self,
    deployment_name_or_id: Union[str, UUID],
    project: Optional[UUID] = None,
) -> DeploymentResponse:
    """Refresh the status of a deployment by name or ID.

    Call this to refresh the operational state of a deployment.

    Args:
        deployment_name_or_id: The name or ID of the deployment to get.
        project: The project ID of the deployment to get. Required
            if a name is provided.

    Returns:
        The deployment.

    Raises:
        DeploymentNotFoundError: if the deployment is not found.
        DeployerError: if an unexpected error occurs.
    """
    client = Client()
    try:
        deployment = client.get_deployment(
            deployment_name_or_id, project=project
        )
    except KeyError:
        raise DeploymentNotFoundError(
            f"Deployment with name or ID '{deployment_name_or_id}' "
            f"not found"
        )

    self._check_deployment_deployer(deployment)

    deployment_state = DeploymentOperationalState(
        status=DeploymentStatus.ERROR,
    )
    try:
        deployment_state = self.do_get_deployment_state(deployment)
    except DeploymentNotFoundError:
        deployment_state.status = DeploymentStatus.ABSENT
    except DeployerError as e:
        raise DeployerError(
            f"Failed to refresh deployment {deployment_name_or_id}: {e}"
        ) from e
    except Exception as e:
        raise DeployerError(
            f"Unexpected error while refreshing deployment for "
            f"{deployment_name_or_id}: {e}"
        ) from e
    finally:
        deployment = self._update_deployment(deployment, deployment_state)

    return deployment

BaseDeployerConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: StackComponentConfig

Base config for all deployers.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)

BaseDeployerFlavor

Bases: Flavor

Base class for deployer flavors.

Attributes

config_class: Type[BaseDeployerConfig] property

Returns BaseDeployerConfig config class.

Returns:

Type	Description
Type[BaseDeployerConfig]	The config class.

implementation_class: Type[BaseDeployer] abstractmethod property

The class that implements the deployer.

type: StackComponentType property

Returns the flavor type.

Returns:

Type	Description
StackComponentType	The flavor type.

ContainerizedDeployer(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Bases: BaseDeployer, ABC

Base class for all containerized deployers.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        environment: Environment variables to set when running on this
            component.
        secrets: Secrets to set as environment variables when running on
            this component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.environment = environment or {}
    self.secrets = secrets or []
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None

Attributes

requirements: Set[str] property

Set of PyPI requirements for the deployer.

Returns:

Type	Description
Set[str]	A set of PyPI requirements for the deployer.

Functions

get_docker_builds(snapshot: PipelineSnapshotBase) -> List[BuildConfiguration]

Gets the Docker builds required for the component.

Parameters:

Name	Type	Description	Default
snapshot	PipelineSnapshotBase	The pipeline snapshot for which to get the builds.	required

Returns:

Type	Description
List[BuildConfiguration]	The required Docker builds.

Source code in src/zenml/deployers/containerized_deployer.py

def get_docker_builds(
    self, snapshot: "PipelineSnapshotBase"
) -> List["BuildConfiguration"]:
    """Gets the Docker builds required for the component.

    Args:
        snapshot: The pipeline snapshot for which to get the builds.

    Returns:
        The required Docker builds.
    """
    return [
        BuildConfiguration(
            key=DEPLOYER_DOCKER_IMAGE_KEY,
            settings=snapshot.pipeline_configuration.docker_settings,
        )
    ]

get_image(snapshot: PipelineSnapshotResponse) -> str staticmethod

Get the docker image used to deploy a pipeline snapshot.

Parameters:

Name	Type	Description	Default
snapshot	PipelineSnapshotResponse	The pipeline snapshot to get the image for.	required

Returns:

Type	Description
str	The docker image used to deploy the pipeline snapshot.

Raises:

Type	Description
RuntimeError	if the pipeline snapshot does not have a build or if the deployer image is not in the build.

Source code in src/zenml/deployers/containerized_deployer.py

@staticmethod
def get_image(snapshot: PipelineSnapshotResponse) -> str:
    """Get the docker image used to deploy a pipeline snapshot.

    Args:
        snapshot: The pipeline snapshot to get the image for.

    Returns:
        The docker image used to deploy the pipeline snapshot.

    Raises:
        RuntimeError: if the pipeline snapshot does not have a build or
            if the deployer image is not in the build.
    """
    if snapshot.build is None:
        raise RuntimeError("Pipeline snapshot does not have a build. ")
    if DEPLOYER_DOCKER_IMAGE_KEY not in snapshot.build.images:
        raise RuntimeError(
            "Pipeline snapshot build does not have a deployer image. "
        )
    return snapshot.build.images[DEPLOYER_DOCKER_IMAGE_KEY].image

DockerDeployer(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Bases: ContainerizedDeployer

Deployer responsible for deploying pipelines locally using Docker.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        environment: Environment variables to set when running on this
            component.
        secrets: Secrets to set as environment variables when running on
            this component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.environment = environment or {}
    self.secrets = secrets or []
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None

Attributes

config: DockerDeployerConfig property

Returns the DockerDeployerConfig config.

Returns:

Type	Description
DockerDeployerConfig	The configuration.

docker_client: DockerClient property

Initialize and/or return the docker client.

Returns:

Type	Description
DockerClient	The docker client.

settings_class: Optional[Type[BaseSettings]] property

Settings class for the Docker deployer.

Returns:

Type	Description
Optional[Type[BaseSettings]]	The settings class.

validator: Optional[StackValidator] property

Ensures there is an image builder in the stack.

Returns:

Type	Description
Optional[StackValidator]	A StackValidator instance.

Functions

do_deprovision_deployment(deployment: DeploymentResponse, timeout: int) -> Optional[DeploymentOperationalState]

Deprovision a docker deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to deprovision.	required
timeout	int	The maximum time in seconds to wait for the pipeline deployment to be deprovisioned.	required

Returns:

Type	Description
Optional[DeploymentOperationalState]	The DeploymentOperationalState object representing the
Optional[DeploymentOperationalState]	operational state of the deleted deployment, or None if the
Optional[DeploymentOperationalState]	deletion is completed before the call returns.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.
DeploymentDeprovisionError	if the deployment deprovision fails.

Source code in src/zenml/deployers/docker/docker_deployer.py

def do_deprovision_deployment(
    self,
    deployment: DeploymentResponse,
    timeout: int,
) -> Optional[DeploymentOperationalState]:
    """Deprovision a docker deployment.

    Args:
        deployment: The deployment to deprovision.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be deprovisioned.

    Returns:
        The DeploymentOperationalState object representing the
        operational state of the deleted deployment, or None if the
        deletion is completed before the call returns.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
        DeploymentDeprovisionError: if the deployment
            deprovision fails.
    """
    container = self._get_container(deployment)
    if container is None:
        raise DeploymentNotFoundError(
            f"Docker container for deployment '{deployment.name}' "
            "not found"
        )

    try:
        container.stop(timeout=timeout)
        container.remove()
    except docker_errors.DockerException as e:
        raise DeploymentDeprovisionError(
            f"Docker container for deployment '{deployment.name}' "
            f"failed to delete: {e}"
        )

    return None

do_get_deployment_state(deployment: DeploymentResponse) -> DeploymentOperationalState

Get information about a docker deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to get information about.	required

Returns:

Type	Description
DeploymentOperationalState	The DeploymentOperationalState object representing the
DeploymentOperationalState	updated operational state of the deployment.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.

Source code in src/zenml/deployers/docker/docker_deployer.py

def do_get_deployment_state(
    self,
    deployment: DeploymentResponse,
) -> DeploymentOperationalState:
    """Get information about a docker deployment.

    Args:
        deployment: The deployment to get information about.

    Returns:
        The DeploymentOperationalState object representing the
        updated operational state of the deployment.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
    """
    container = self._get_container(deployment)
    if container is None:
        raise DeploymentNotFoundError(
            f"Docker container for deployment '{deployment.name}' "
            "not found"
        )

    return self._get_container_operational_state(container)

do_get_deployment_state_logs(deployment: DeploymentResponse, follow: bool = False, tail: Optional[int] = None) -> Generator[str, bool, None]

Get the logs of a Docker deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to get the logs of.	required
follow	bool	if True, the logs will be streamed as they are written	False
tail	Optional[int]	only retrieve the last NUM lines of log output.	None

Yields:

Type	Description
str	The logs of the deployment.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.
DeploymentLogsNotFoundError	if the deployment logs are not found.
DeployerError	if the deployment logs cannot be retrieved for any other reason or if an unexpected error occurs.

Source code in src/zenml/deployers/docker/docker_deployer.py

def do_get_deployment_state_logs(
    self,
    deployment: DeploymentResponse,
    follow: bool = False,
    tail: Optional[int] = None,
) -> Generator[str, bool, None]:
    """Get the logs of a Docker deployment.

    Args:
        deployment: The deployment to get the logs of.
        follow: if True, the logs will be streamed as they are written
        tail: only retrieve the last NUM lines of log output.

    Yields:
        The logs of the deployment.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
        DeploymentLogsNotFoundError: if the deployment logs are not
            found.
        DeployerError: if the deployment logs cannot
            be retrieved for any other reason or if an unexpected error
            occurs.
    """
    container = self._get_container(deployment)
    if container is None:
        raise DeploymentNotFoundError(
            f"Docker container for deployment '{deployment.name}' "
            "not found"
        )

    try:
        log_kwargs: Dict[str, Any] = {
            "stdout": True,
            "stderr": True,
            "stream": follow,
            "follow": follow,
            "timestamps": True,
        }

        if tail is not None and tail > 0:
            log_kwargs["tail"] = tail

        log_stream = container.logs(**log_kwargs)

        if follow:
            for log_line in log_stream:
                if isinstance(log_line, bytes):
                    yield log_line.decode(
                        "utf-8", errors="replace"
                    ).rstrip()
                else:
                    yield str(log_line).rstrip()
        else:
            if isinstance(log_stream, bytes):
                log_text = log_stream.decode("utf-8", errors="replace")
                for line in log_text.splitlines():
                    yield line
            else:
                for log_line in log_stream:
                    if isinstance(log_line, bytes):
                        yield log_line.decode(
                            "utf-8", errors="replace"
                        ).rstrip()
                    else:
                        yield str(log_line).rstrip()

    except docker_errors.NotFound as e:
        raise DeploymentLogsNotFoundError(
            f"Logs for deployment '{deployment.name}' not found: {e}"
        )
    except docker_errors.APIError as e:
        raise DeployerError(
            f"Docker API error while retrieving logs for deployment "
            f"'{deployment.name}': {e}"
        )
    except docker_errors.DockerException as e:
        raise DeployerError(
            f"Docker error while retrieving logs for deployment "
            f"'{deployment.name}': {e}"
        )
    except Exception as e:
        raise DeployerError(
            f"Unexpected error while retrieving logs for deployment "
            f"'{deployment.name}': {e}"
        )

do_provision_deployment(deployment: DeploymentResponse, stack: Stack, environment: Dict[str, str], secrets: Dict[str, str], timeout: int) -> DeploymentOperationalState

Deploy a pipeline as a Docker container.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to run as a Docker container.	required
stack	Stack	The stack the pipeline will be deployed on.	required
environment	Dict[str, str]	A dictionary of environment variables to set on the deployment.	required
secrets	Dict[str, str]	A dictionary of secret environment variables to set on the deployment. These secret environment variables should not be exposed as regular environment variables on the deployer.	required
timeout	int	The maximum time in seconds to wait for the pipeline deployment to be provisioned.	required

Returns:

Type	Description
DeploymentOperationalState	The DeploymentOperationalState object representing the
DeploymentOperationalState	operational state of the provisioned deployment.

Raises:

Type	Description
DeploymentProvisionError	if provisioning the deployment fails.

Source code in src/zenml/deployers/docker/docker_deployer.py

def do_provision_deployment(
    self,
    deployment: DeploymentResponse,
    stack: "Stack",
    environment: Dict[str, str],
    secrets: Dict[str, str],
    timeout: int,
) -> DeploymentOperationalState:
    """Deploy a pipeline as a Docker container.

    Args:
        deployment: The deployment to run as a Docker container.
        stack: The stack the pipeline will be deployed on.
        environment: A dictionary of environment variables to set on the
            deployment.
        secrets: A dictionary of secret environment variables to set
            on the deployment. These secret environment variables
            should not be exposed as regular environment variables on the
            deployer.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be provisioned.

    Returns:
        The DeploymentOperationalState object representing the
        operational state of the provisioned deployment.

    Raises:
        DeploymentProvisionError: if provisioning the deployment
            fails.
    """
    assert deployment.snapshot, "Pipeline snapshot not found"
    snapshot = deployment.snapshot

    # Currently, there is no safe way to pass secrets to a docker
    # container, so we simply merge them into the environment variables.
    environment.update(secrets)

    settings = cast(
        DockerDeployerSettings,
        self.get_settings(snapshot),
    )

    existing_metadata = DockerDeploymentMetadata.from_deployment(
        deployment
    )

    entrypoint = DeploymentEntrypointConfiguration.get_entrypoint_command()

    entrypoint_kwargs = {
        DEPLOYMENT_ID_OPTION: deployment.id,
        PORT_OPTION: 8000,
    }
    if deployment.auth_key:
        entrypoint_kwargs[AUTH_KEY_OPTION] = deployment.auth_key

    arguments = DeploymentEntrypointConfiguration.get_entrypoint_arguments(
        **entrypoint_kwargs
    )

    # Add the local stores path as a volume mount
    stack.check_local_paths()
    local_stores_path = GlobalConfiguration().local_stores_path
    volumes = {
        local_stores_path: {
            "bind": local_stores_path,
            "mode": "rw",
        }
    }
    environment[ENV_ZENML_LOCAL_STORES_PATH] = local_stores_path

    # check if a container already exists for the deployment
    container = self._get_container(deployment)

    if container:
        # the container exists, check if it is running
        if container.status == "running":
            logger.debug(
                f"Container for deployment '{deployment.name}' is "
                "already running",
            )
            container.stop(timeout=timeout)

        # the container is stopped or in an error state, remove it
        logger.debug(
            f"Removing previous container for deployment "
            f"'{deployment.name}'",
        )
        container.remove(force=True)

    logger.debug(
        f"Starting container for deployment '{deployment.name}'..."
    )

    image = self.get_image(deployment.snapshot)

    try:
        self.docker_client.images.get(image)
    except docker_errors.ImageNotFound:
        logger.debug(
            f"Pulling container image '{image}' for deployment "
            f"'{deployment.name}'...",
        )
        self.docker_client.images.pull(image)

    preferred_ports: List[int] = []
    if settings.port:
        preferred_ports.append(settings.port)
    if existing_metadata.port:
        preferred_ports.append(existing_metadata.port)
    port = lookup_preferred_or_free_port(
        preferred_ports=preferred_ports,
        allocate_port_if_busy=settings.allocate_port_if_busy,
        range=settings.port_range,
    )
    ports: Dict[str, Optional[int]] = {"8000/tcp": port}

    uid_args: Dict[str, Any] = {}
    if sys.platform == "win32":
        # File permissions are not checked on Windows. This if clause
        # prevents mypy from complaining about unused 'type: ignore'
        # statements
        pass
    else:
        # Run the container in the context of the local UID/GID
        # to ensure that the local database can be shared
        # with the container.
        logger.debug(
            "Setting UID and GID to local user/group in container."
        )
        uid_args = dict(
            user=os.getuid(),
            group_add=[os.getgid()],
        )

    run_args = copy.deepcopy(settings.run_args)
    docker_environment = run_args.pop("environment", {})
    docker_environment.update(environment)

    docker_volumes = run_args.pop("volumes", {})
    docker_volumes.update(volumes)

    extra_hosts = run_args.pop("extra_hosts", {})
    extra_hosts["host.docker.internal"] = "host-gateway"

    run_args.update(uid_args)

    try:
        container = self.docker_client.containers.run(
            image=image,
            name=self._get_container_id(deployment),
            entrypoint=entrypoint,
            command=arguments,
            detach=True,
            volumes=docker_volumes,
            environment=docker_environment,
            remove=False,
            auto_remove=False,
            ports=ports,
            labels={
                "zenml-deployment-id": str(deployment.id),
                "zenml-deployment-name": deployment.name,
                "zenml-deployer-name": str(self.name),
                "zenml-deployer-id": str(self.id),
                "managed-by": "zenml",
            },
            extra_hosts=extra_hosts,
            **run_args,
        )

        logger.debug(
            f"Docker container for deployment '{deployment.name}' "
            f"started with ID {self._get_container_id(deployment)}",
        )

    except docker_errors.DockerException as e:
        raise DeploymentProvisionError(
            f"Docker container for deployment '{deployment.name}' "
            f"failed to start: {e}"
        )

    return self._get_container_operational_state(container)

DockerDeployerFlavor

Bases: BaseDeployerFlavor

Flavor for the Docker deployer.

Attributes

config_class: Type[BaseDeployerConfig] property

Config class for the base deployer flavor.

Returns:

Type	Description
Type[BaseDeployerConfig]	The config class.

docs_url: Optional[str] property

A url to point at docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor docs url.

implementation_class: Type[DockerDeployer] property

Implementation class for this flavor.

Returns:

Type	Description
Type[DockerDeployer]	Implementation class for this flavor.

logo_url: str property

A url to represent the flavor in the dashboard.

Returns:

Type	Description
str	The flavor logo.

name: str property

Name of the deployer flavor.

Returns:

Type	Description
str	Name of the deployer flavor.

sdk_docs_url: Optional[str] property

A url to point at SDK docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor SDK docs url.

Modules

base_deployer

Base class for all ZenML deployers.

Classes

BaseDeployer(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Bases: StackComponent, ABC

Base class for all ZenML deployers.

The deployer serves three major purposes:

1. It contains all the stack related configuration attributes required to interact with the remote pipeline deployment tool, service or platform (e.g. hostnames, URLs, references to credentials, other client related configuration parameters).

2. It implements the life-cycle management for deployments, including discovery, creation, deletion and updating.

3. It acts as a ZenML deployment registry, where every pipeline deployment is stored as a database entity through the ZenML Client. This allows the deployer to keep track of all externally running pipeline deployments and to manage their lifecycle.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        environment: Environment variables to set when running on this
            component.
        secrets: Secrets to set as environment variables when running on
            this component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.environment = environment or {}
    self.secrets = secrets or []
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None

Attributes

config: BaseDeployerConfig property

Returns the BaseDeployerConfig config.

Returns:

Type	Description
BaseDeployerConfig	The configuration.

Functions

delete_deployment(deployment_name_or_id: Union[str, UUID], project: Optional[UUID] = None, force: bool = False, timeout: Optional[int] = None) -> None

Deprovision and delete a deployment.

Parameters:

Name	Type	Description	Default
deployment_name_or_id	Union[str, UUID]	The name or ID of the deployment to delete.	required
project	Optional[UUID]	The project ID of the deployment to deprovision. Required if a name is provided.	None
force	bool	if True, force the deployment to delete even if it cannot be deprovisioned.	False
timeout	Optional[int]	The maximum time in seconds to wait for the pipeline deployment to be deprovisioned. If provided, will override the deployer's default timeout.	None

Raises:

Type	Description
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

def delete_deployment(
    self,
    deployment_name_or_id: Union[str, UUID],
    project: Optional[UUID] = None,
    force: bool = False,
    timeout: Optional[int] = None,
) -> None:
    """Deprovision and delete a deployment.

    Args:
        deployment_name_or_id: The name or ID of the deployment to
            delete.
        project: The project ID of the deployment to deprovision.
            Required if a name is provided.
        force: if True, force the deployment to delete even if it
            cannot be deprovisioned.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be deprovisioned. If provided, will override the
            deployer's default timeout.

    Raises:
        DeployerError: if an unexpected error occurs.
    """
    client = Client()
    try:
        deployment = self.deprovision_deployment(
            deployment_name_or_id, project, timeout
        )
    except DeploymentNotFoundError:
        # The deployment was already deleted
        return
    except DeployerError as e:
        if force:
            logger.warning(
                f"Failed to deprovision deployment "
                f"{deployment_name_or_id}: {e}. Forcing deletion."
            )
            deployment = client.get_deployment(
                deployment_name_or_id, project=project
            )
            client.zen_store.delete_deployment(deployment_id=deployment.id)
        else:
            raise
    else:
        client.zen_store.delete_deployment(deployment_id=deployment.id)

deprovision_deployment(deployment_name_or_id: Union[str, UUID], project: Optional[UUID] = None, timeout: Optional[int] = None) -> DeploymentResponse

Deprovision a deployment.

Parameters:

Name	Type	Description	Default
deployment_name_or_id	Union[str, UUID]	The name or ID of the deployment to deprovision.	required
project	Optional[UUID]	The project ID of the deployment to deprovision. Required if a name is provided.	None
timeout	Optional[int]	The maximum time in seconds to wait for the pipeline deployment to deprovision. If provided, will override the deployer's default timeout.	None

Returns:

Type	Description
DeploymentResponse	The deployment.

Raises:

Type	Description
DeploymentNotFoundError	if the deployment is not found or is not managed by this deployer.
DeploymentDeprovisionError	if the deployment deprovision fails.
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

def deprovision_deployment(
    self,
    deployment_name_or_id: Union[str, UUID],
    project: Optional[UUID] = None,
    timeout: Optional[int] = None,
) -> DeploymentResponse:
    """Deprovision a deployment.

    Args:
        deployment_name_or_id: The name or ID of the deployment to
            deprovision.
        project: The project ID of the deployment to deprovision.
            Required if a name is provided.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to deprovision. If provided, will override the
            deployer's default timeout.

    Returns:
        The deployment.

    Raises:
        DeploymentNotFoundError: if the deployment is not found
            or is not managed by this deployer.
        DeploymentDeprovisionError: if the deployment
            deprovision fails.
        DeployerError: if an unexpected error occurs.
    """
    client = Client()
    try:
        deployment = client.get_deployment(
            deployment_name_or_id, project=project
        )
    except KeyError:
        raise DeploymentNotFoundError(
            f"Deployment with name or ID '{deployment_name_or_id}' "
            f"not found"
        )

    self._check_deployment_deployer(deployment)

    if not timeout and deployment.snapshot:
        settings = cast(
            BaseDeployerSettings,
            self.get_settings(deployment.snapshot),
        )

        timeout = settings.lcm_timeout

    timeout = timeout or DEFAULT_DEPLOYMENT_LCM_TIMEOUT

    start_time = time.time()
    deployment_state = DeploymentOperationalState(
        status=DeploymentStatus.ERROR,
    )
    with track_handler(
        AnalyticsEvent.STOP_DEPLOYMENT
    ) as analytics_handler:
        try:
            deleted_deployment_state = self.do_deprovision_deployment(
                deployment, timeout
            )
            if not deleted_deployment_state:
                # When do_delete_deployment returns a None value, this
                # is to signal that the deployment is already fully deprovisioned.
                deployment_state.status = DeploymentStatus.ABSENT
        except DeploymentNotFoundError:
            deployment_state.status = DeploymentStatus.ABSENT
        except DeployerError as e:
            raise DeployerError(
                f"Failed to delete deployment {deployment_name_or_id}: {e}"
            ) from e
        except Exception as e:
            raise DeployerError(
                f"Unexpected error while deleting deployment for "
                f"{deployment_name_or_id}: {e}"
            ) from e
        finally:
            deployment = self._update_deployment(
                deployment, deployment_state
            )

        try:
            if deployment_state.status == DeploymentStatus.ABSENT:
                return deployment

            # Subtract the time spent deprovisioning the deployment from the timeout
            timeout = timeout - int(time.time() - start_time)
            deployment, _ = self._poll_deployment(
                deployment, DeploymentStatus.ABSENT, timeout
            )

            if deployment.status != DeploymentStatus.ABSENT:
                raise DeploymentDeprovisionError(
                    f"Failed to deprovision deployment {deployment_name_or_id}: "
                    f"Operational state: {deployment.status}"
                )

        finally:
            analytics_handler.metadata = (
                self._get_deployment_analytics_metadata(
                    deployment=deployment,
                    stack=None,
                )
            )

        return deployment

do_deprovision_deployment(deployment: DeploymentResponse, timeout: int) -> Optional[DeploymentOperationalState] abstractmethod

Abstract method to deprovision a deployment.

Concrete deployer subclasses must implement the following functionality in this method:

• Deprovision the actual deployment infrastructure (e.g., FastAPI server, Kubernetes deployment, cloud function, etc.) based on the information in the deployment response.

• Return a DeploymentOperationalState representing the operational state of the deleted deployment, or None if the deletion is completed before the call returns.

Note that the deployment infrastructure is not required to be deleted immediately. The deployer can return a DeploymentOperationalState with a status of DeploymentStatus.PENDING, and the base deployer will poll the deployment infrastructure by calling the do_get_deployment_state method until it is deleted or it times out.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to delete.	required
timeout	int	The maximum time in seconds to wait for the pipeline deployment to be deprovisioned.	required

Returns:

Type	Description
Optional[DeploymentOperationalState]	The DeploymentOperationalState object representing the
Optional[DeploymentOperationalState]	operational state of the deprovisioned deployment, or None
Optional[DeploymentOperationalState]	if the deprovision is completed before the call returns.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.
DeploymentDeprovisionError	if the deployment deprovision fails.
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

@abstractmethod
def do_deprovision_deployment(
    self,
    deployment: DeploymentResponse,
    timeout: int,
) -> Optional[DeploymentOperationalState]:
    """Abstract method to deprovision a deployment.

    Concrete deployer subclasses must implement the following
    functionality in this method:

    - Deprovision the actual deployment infrastructure (e.g.,
    FastAPI server, Kubernetes deployment, cloud function, etc.) based on
    the information in the deployment response.

    - Return a DeploymentOperationalState representing the operational
    state of the deleted deployment, or None if the deletion is
    completed before the call returns.

    Note that the deployment infrastructure is not required to be
    deleted immediately. The deployer can return a
    DeploymentOperationalState with a status of
    DeploymentStatus.PENDING, and the base deployer will poll
    the deployment infrastructure by calling the
    `do_get_deployment_state` method until it is deleted or it times out.

    Args:
        deployment: The deployment to delete.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be deprovisioned.

    Returns:
        The DeploymentOperationalState object representing the
        operational state of the deprovisioned deployment, or None
        if the deprovision is completed before the call returns.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
        DeploymentDeprovisionError: if the deployment
            deprovision fails.
        DeployerError: if an unexpected error occurs.
    """

do_get_deployment_state(deployment: DeploymentResponse) -> DeploymentOperationalState abstractmethod

Abstract method to get information about a deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to get information about.	required

Returns:

Type	Description
DeploymentOperationalState	The DeploymentOperationalState object representing the
DeploymentOperationalState	updated operational state of the deployment.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.
DeployerError	if the deployment information cannot be retrieved for any other reason or if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

@abstractmethod
def do_get_deployment_state(
    self,
    deployment: DeploymentResponse,
) -> DeploymentOperationalState:
    """Abstract method to get information about a deployment.

    Args:
        deployment: The deployment to get information about.

    Returns:
        The DeploymentOperationalState object representing the
        updated operational state of the deployment.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
        DeployerError: if the deployment information cannot
            be retrieved for any other reason or if an unexpected error
            occurs.
    """

do_get_deployment_state_logs(deployment: DeploymentResponse, follow: bool = False, tail: Optional[int] = None) -> Generator[str, bool, None] abstractmethod

Abstract method to get the logs of a deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to get the logs of.	required
follow	bool	if True, the logs will be streamed as they are written	False
tail	Optional[int]	only retrieve the last NUM lines of log output.	None

Yields:

Type	Description
str	The logs of the deployment.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.
DeploymentLogsNotFoundError	if the deployment logs are not found.
DeployerError	if the deployment logs cannot be retrieved for any other reason or if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

@abstractmethod
def do_get_deployment_state_logs(
    self,
    deployment: DeploymentResponse,
    follow: bool = False,
    tail: Optional[int] = None,
) -> Generator[str, bool, None]:
    """Abstract method to get the logs of a deployment.

    Args:
        deployment: The deployment to get the logs of.
        follow: if True, the logs will be streamed as they are written
        tail: only retrieve the last NUM lines of log output.

    Yields:
        The logs of the deployment.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
        DeploymentLogsNotFoundError: if the deployment logs are not
            found.
        DeployerError: if the deployment logs cannot
            be retrieved for any other reason or if an unexpected error
            occurs.
    """

do_provision_deployment(deployment: DeploymentResponse, stack: Stack, environment: Dict[str, str], secrets: Dict[str, str], timeout: int) -> DeploymentOperationalState abstractmethod

Abstract method to deploy a pipeline as an HTTP deployment.

Concrete deployer subclasses must implement the following functionality in this method:

• Create the actual deployment infrastructure (e.g., FastAPI server, Kubernetes deployment, cloud function, etc.) based on the information in the deployment response, particularly the pipeline snapshot. When determining how to name the external resources, do not rely on the deployment name as being immutable or unique.

• If the deployment infrastructure is already provisioned, update it to match the information in the deployment response.

• Return a DeploymentOperationalState representing the operational state of the provisioned deployment.

Note that the deployment infrastructure is not required to be deployed immediately. The deployer can return a DeploymentOperationalState with a status of DeploymentStatus.PENDING, and the base deployer will poll the deployment infrastructure by calling the do_get_deployment_state method until it is ready or it times out.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to deploy as an HTTP deployment.	required
stack	Stack	The stack the pipeline will be deployed on.	required
environment	Dict[str, str]	A dictionary of environment variables to set on the deployment.	required
secrets	Dict[str, str]	A dictionary of secret environment variables to set on the deployment. These secret environment variables should not be exposed as regular environment variables on the deployer.	required
timeout	int	The maximum time in seconds to wait for the pipeline deployment to be provisioned.	required

Returns:

Type	Description
DeploymentOperationalState	The DeploymentOperationalState object representing the
DeploymentOperationalState	operational state of the provisioned deployment.

Raises:

Type	Description
DeploymentProvisionError	if provisioning the deployment fails.
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

@abstractmethod
def do_provision_deployment(
    self,
    deployment: DeploymentResponse,
    stack: "Stack",
    environment: Dict[str, str],
    secrets: Dict[str, str],
    timeout: int,
) -> DeploymentOperationalState:
    """Abstract method to deploy a pipeline as an HTTP deployment.

    Concrete deployer subclasses must implement the following
    functionality in this method:

    - Create the actual deployment infrastructure (e.g.,
    FastAPI server, Kubernetes deployment, cloud function, etc.) based on
    the information in the deployment response, particularly the
    pipeline snapshot. When determining how to name the external
    resources, do not rely on the deployment name as being immutable
    or unique.

    - If the deployment infrastructure is already provisioned, update
    it to match the information in the deployment response.

    - Return a DeploymentOperationalState representing the operational
    state of the provisioned deployment.

    Note that the deployment infrastructure is not required to be
    deployed immediately. The deployer can return a
    DeploymentOperationalState with a status of
    DeploymentStatus.PENDING, and the base deployer will poll
    the deployment infrastructure by calling the
    `do_get_deployment_state` method until it is ready or it times out.

    Args:
        deployment: The deployment to deploy as an HTTP deployment.
        stack: The stack the pipeline will be deployed on.
        environment: A dictionary of environment variables to set on the
            deployment.
        secrets: A dictionary of secret environment variables to set
            on the deployment. These secret environment variables
            should not be exposed as regular environment variables on the
            deployer.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be provisioned.

    Returns:
        The DeploymentOperationalState object representing the
        operational state of the provisioned deployment.

    Raises:
        DeploymentProvisionError: if provisioning the deployment
            fails.
        DeployerError: if an unexpected error occurs.
    """

get_active_deployer() -> BaseDeployer classmethod

Get the deployer registered in the active stack.

Returns:

Type	Description
BaseDeployer	The deployer registered in the active stack.

Raises:

Type	Description
TypeError	if a deployer is not part of the active stack.

Source code in src/zenml/deployers/base_deployer.py

@classmethod
def get_active_deployer(cls) -> "BaseDeployer":
    """Get the deployer registered in the active stack.

    Returns:
        The deployer registered in the active stack.

    Raises:
        TypeError: if a deployer is not part of the
            active stack.
    """
    client = Client()
    deployer = client.active_stack.deployer
    if not deployer or not isinstance(deployer, cls):
        raise TypeError(
            "The active stack needs to have a deployer "
            "component registered to be able to deploy pipelines. "
            "You can create a new stack with a deployer component "
            "or update your active stack to add this component, e.g.:\n\n"
            "  `zenml deployer register ...`\n"
            "  `zenml stack register <STACK-NAME> -D ...`\n"
            "  or:\n"
            "  `zenml stack update -D ...`\n\n"
        )

    return deployer

get_deployment_logs(deployment_name_or_id: Union[str, UUID], project: Optional[UUID] = None, follow: bool = False, tail: Optional[int] = None) -> Generator[str, bool, None]

Get the logs of a deployment.

Parameters:

Name	Type	Description	Default
deployment_name_or_id	Union[str, UUID]	The name or ID of the deployment to get the logs of.	required
project	Optional[UUID]	The project ID of the deployment to get the logs of. Required if a name is provided.	None
follow	bool	if True, the logs will be streamed as they are written.	False
tail	Optional[int]	only retrieve the last NUM lines of log output.	None

Returns:

Type	Description
None	A generator that yields the logs of the deployment.

Raises:

Type	Description
DeploymentNotFoundError	if the deployment is not found.
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

def get_deployment_logs(
    self,
    deployment_name_or_id: Union[str, UUID],
    project: Optional[UUID] = None,
    follow: bool = False,
    tail: Optional[int] = None,
) -> Generator[str, bool, None]:
    """Get the logs of a deployment.

    Args:
        deployment_name_or_id: The name or ID of the deployment to get
            the logs of.
        project: The project ID of the deployment to get the logs of.
            Required if a name is provided.
        follow: if True, the logs will be streamed as they are written.
        tail: only retrieve the last NUM lines of log output.

    Returns:
        A generator that yields the logs of the deployment.

    Raises:
        DeploymentNotFoundError: if the deployment is not found.
        DeployerError: if an unexpected error occurs.
    """
    client = Client()
    try:
        deployment = client.get_deployment(
            deployment_name_or_id, project=project
        )
    except KeyError:
        raise DeploymentNotFoundError(
            f"Deployment with name or ID '{deployment_name_or_id}' "
            f"not found"
        )

    self._check_deployment_deployer(deployment)

    try:
        return self.do_get_deployment_state_logs(deployment, follow, tail)
    except DeployerError as e:
        raise DeployerError(
            f"Failed to get logs for deployment {deployment_name_or_id}: {e}"
        ) from e
    except Exception as e:
        raise DeployerError(
            f"Unexpected error while getting logs for deployment for "
            f"{deployment_name_or_id}: {e}"
        ) from e

provision_deployment(snapshot: PipelineSnapshotResponse, stack: Stack, deployment_name_or_id: Union[str, UUID], replace: bool = True, timeout: Optional[int] = None) -> DeploymentResponse

Provision a deployment.

The provision_deployment method is the main entry point for provisioning deployments using the deployer. It is used to deploy a pipeline snapshot as an HTTP deployment, or update an existing deployment instance with the same name. The method returns a DeploymentResponse object that is a representation of the external deployment instance.

Parameters:

Name	Type	Description	Default
snapshot	PipelineSnapshotResponse	The pipeline snapshot to deploy as an HTTP deployment.	required
stack	Stack	The stack the pipeline will be deployed on.	required
deployment_name_or_id	Union[str, UUID]	Unique name or ID for the deployment. This name must be unique at the project level.	required
replace	bool	If True, it will update in-place any existing pipeline deployment instance with the same name. If False, and the pipeline deployment instance already exists, it will raise a DeploymentAlreadyExistsError.	True
timeout	Optional[int]	The maximum time in seconds to wait for the pipeline deployment to be provisioned. If provided, will override the deployer's default timeout.	None

Raises:

Type	Description
DeploymentAlreadyExistsError	if the deployment already exists and replace is False.
DeploymentProvisionError	if the deployment fails.
DeploymentSnapshotMismatchError	if the pipeline snapshot was not created for this deployer.
DeploymentNotFoundError	if the deployment with the given ID is not found.
DeployerError	if an unexpected error occurs.

Returns:

Type	Description
DeploymentResponse	The DeploymentResponse object representing the provisioned
DeploymentResponse	deployment.

Source code in src/zenml/deployers/base_deployer.py

def provision_deployment(
    self,
    snapshot: PipelineSnapshotResponse,
    stack: "Stack",
    deployment_name_or_id: Union[str, UUID],
    replace: bool = True,
    timeout: Optional[int] = None,
) -> DeploymentResponse:
    """Provision a deployment.

    The provision_deployment method is the main entry point for
    provisioning deployments using the deployer. It is used to deploy
    a pipeline snapshot as an HTTP deployment, or update an existing
    deployment instance with the same name. The method returns a
    DeploymentResponse object that is a representation of the
    external deployment instance.

    Args:
        snapshot: The pipeline snapshot to deploy as an HTTP deployment.
        stack: The stack the pipeline will be deployed on.
        deployment_name_or_id: Unique name or ID for the deployment.
            This name must be unique at the project level.
        replace: If True, it will update in-place any existing pipeline
            deployment instance with the same name. If False, and the pipeline
            deployment instance already exists, it will raise a
            DeploymentAlreadyExistsError.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be provisioned. If provided, will override the
            deployer's default timeout.

    Raises:
        DeploymentAlreadyExistsError: if the deployment already
            exists and replace is False.
        DeploymentProvisionError: if the deployment fails.
        DeploymentSnapshotMismatchError: if the pipeline snapshot
            was not created for this deployer.
        DeploymentNotFoundError: if the deployment with the
            given ID is not found.
        DeployerError: if an unexpected error occurs.

    Returns:
        The DeploymentResponse object representing the provisioned
        deployment.
    """
    if not replace and is_valid_uuid(deployment_name_or_id):
        raise DeploymentAlreadyExistsError(
            f"A deployment with ID '{deployment_name_or_id}' "
            "already exists"
        )

    self._check_deployment_inputs_outputs(snapshot)

    client = Client()

    settings = cast(
        BaseDeployerSettings,
        self.get_settings(snapshot),
    )

    timeout = timeout or settings.lcm_timeout
    auth_key = settings.auth_key
    if not auth_key and settings.generate_auth_key:
        auth_key = self._generate_auth_key()

    if snapshot.stack and snapshot.stack.id != stack.id:
        # When a different stack is used then the one the snapshot was
        # created for, the container image may not have the correct
        # dependencies installed, which leads to unexpected errors during
        # deployment. To avoid this, we raise an error here.
        raise DeploymentSnapshotMismatchError(
            f"The pipeline snapshot with ID '{snapshot.id}' "
            f"was not created for the stack {stack.name} and might not "
            "have the correct dependencies installed. This may "
            "lead to unexpected behavior during deployment. Please switch "
            f"to the correct active stack '{snapshot.stack.name}' or use "
            "a different snapshot."
        )

    try:
        # Get the existing deployment
        deployment = client.get_deployment(
            deployment_name_or_id, project=snapshot.project_id
        )

        self._check_snapshot_already_deployed(snapshot, deployment.id)

        logger.debug(
            f"Existing deployment found with name '{deployment.name}'"
        )
    except KeyError:
        if isinstance(deployment_name_or_id, UUID):
            raise DeploymentNotFoundError(
                f"Deployment with ID '{deployment_name_or_id}' not found"
            )

        self._check_snapshot_already_deployed(
            snapshot, deployment_name_or_id
        )

        logger.debug(
            f"Creating new deployment {deployment_name_or_id} with "
            f"snapshot ID: {snapshot.id}"
        )

        # Create the deployment request
        deployment_request = DeploymentRequest(
            name=deployment_name_or_id,
            project=snapshot.project_id,
            snapshot_id=snapshot.id,
            deployer_id=self.id,  # This deployer's ID
            auth_key=auth_key,
        )

        deployment = client.zen_store.create_deployment(deployment_request)
        logger.debug(
            f"Created new deployment with name '{deployment.name}' "
            f"and ID: {deployment.id}"
        )
    else:
        if not replace:
            raise DeploymentAlreadyExistsError(
                f"A deployment with name '{deployment.name}' "
                "already exists"
            )

        self._check_deployment_deployer(deployment)
        self._check_deployment_snapshot(snapshot)

        deployment_update = DeploymentUpdate(
            snapshot_id=snapshot.id,
        )
        if (
            deployment.auth_key
            and not auth_key
            or not deployment.auth_key
            and auth_key
        ):
            # Key was either added or removed
            deployment_update.auth_key = auth_key
        elif deployment.auth_key != auth_key and (
            settings.auth_key or not settings.generate_auth_key
        ):
            # Key was changed and not because of re-generation
            deployment_update.auth_key = auth_key

        # The deployment has been updated
        deployment = client.zen_store.update_deployment(
            deployment.id,
            deployment_update,
        )

    logger.info(
        f"Provisioning deployment {deployment.name} with "
        f"snapshot ID: {snapshot.id}"
    )

    environment, secrets = get_config_environment_vars(
        deployment_id=deployment.id,
    )

    # Make sure to use the correct active stack/project which correspond
    # to the supplied stack and snapshot, which may be different from the
    # active stack/project
    environment[ENV_ZENML_ACTIVE_STACK_ID] = str(stack.id)
    environment[ENV_ZENML_ACTIVE_PROJECT_ID] = str(snapshot.project_id)

    start_time = time.time()
    deployment_state = DeploymentOperationalState(
        status=DeploymentStatus.ERROR,
    )
    with track_handler(
        AnalyticsEvent.DEPLOY_PIPELINE
    ) as analytics_handler:
        try:
            deployment_state = self.do_provision_deployment(
                deployment,
                stack=stack,
                environment=environment,
                secrets=secrets,
                timeout=timeout,
            )
        except DeploymentProvisionError as e:
            raise DeploymentProvisionError(
                f"Failed to provision deployment {deployment.name}: {e}"
            ) from e
        except DeployerError as e:
            raise DeployerError(
                f"Failed to provision deployment {deployment.name}: {e}"
            ) from e
        except Exception as e:
            raise DeployerError(
                f"Unexpected error while provisioning deployment for "
                f"{deployment.name}: {e}"
            ) from e
        finally:
            deployment = self._update_deployment(
                deployment, deployment_state
            )

        logger.info(
            f"Provisioned deployment {deployment.name} with "
            f"snapshot ID: {snapshot.id}. Operational state is: "
            f"{deployment_state.status}"
        )

        try:
            if deployment_state.status == DeploymentStatus.RUNNING:
                return deployment

            # Subtract the time spent deploying the deployment from the
            # timeout
            timeout = timeout - int(time.time() - start_time)
            deployment, _ = self._poll_deployment(
                deployment, DeploymentStatus.RUNNING, timeout
            )

            if deployment.status != DeploymentStatus.RUNNING:
                raise DeploymentProvisionError(
                    f"Failed to provision deployment {deployment.name}: "
                    f"The deployment's operational state is "
                    f"{deployment.status}. Please check the status or logs "
                    "of the deployment for more information."
                )

        finally:
            analytics_handler.metadata = (
                self._get_deployment_analytics_metadata(
                    deployment=deployment,
                    stack=stack,
                )
            )

        return deployment

refresh_deployment(deployment_name_or_id: Union[str, UUID], project: Optional[UUID] = None) -> DeploymentResponse

Refresh the status of a deployment by name or ID.

Call this to refresh the operational state of a deployment.

Parameters:

Name	Type	Description	Default
deployment_name_or_id	Union[str, UUID]	The name or ID of the deployment to get.	required
project	Optional[UUID]	The project ID of the deployment to get. Required if a name is provided.	None

Returns:

Type	Description
DeploymentResponse	The deployment.

Raises:

Type	Description
DeploymentNotFoundError	if the deployment is not found.
DeployerError	if an unexpected error occurs.

Source code in src/zenml/deployers/base_deployer.py

def refresh_deployment(
    self,
    deployment_name_or_id: Union[str, UUID],
    project: Optional[UUID] = None,
) -> DeploymentResponse:
    """Refresh the status of a deployment by name or ID.

    Call this to refresh the operational state of a deployment.

    Args:
        deployment_name_or_id: The name or ID of the deployment to get.
        project: The project ID of the deployment to get. Required
            if a name is provided.

    Returns:
        The deployment.

    Raises:
        DeploymentNotFoundError: if the deployment is not found.
        DeployerError: if an unexpected error occurs.
    """
    client = Client()
    try:
        deployment = client.get_deployment(
            deployment_name_or_id, project=project
        )
    except KeyError:
        raise DeploymentNotFoundError(
            f"Deployment with name or ID '{deployment_name_or_id}' "
            f"not found"
        )

    self._check_deployment_deployer(deployment)

    deployment_state = DeploymentOperationalState(
        status=DeploymentStatus.ERROR,
    )
    try:
        deployment_state = self.do_get_deployment_state(deployment)
    except DeploymentNotFoundError:
        deployment_state.status = DeploymentStatus.ABSENT
    except DeployerError as e:
        raise DeployerError(
            f"Failed to refresh deployment {deployment_name_or_id}: {e}"
        ) from e
    except Exception as e:
        raise DeployerError(
            f"Unexpected error while refreshing deployment for "
            f"{deployment_name_or_id}: {e}"
        ) from e
    finally:
        deployment = self._update_deployment(deployment, deployment_state)

    return deployment

BaseDeployerConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: StackComponentConfig

Base config for all deployers.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)

BaseDeployerFlavor

Bases: Flavor

Base class for deployer flavors.

Attributes

config_class: Type[BaseDeployerConfig] property

Returns BaseDeployerConfig config class.

Returns:

Type	Description
Type[BaseDeployerConfig]	The config class.

implementation_class: Type[BaseDeployer] abstractmethod property

The class that implements the deployer.

type: StackComponentType property

Returns the flavor type.

Returns:

Type	Description
StackComponentType	The flavor type.

BaseDeployerSettings(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseSettings

Base settings for all deployers.

Source code in src/zenml/config/secret_reference_mixin.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references are only passed for valid fields.

    This method ensures that secret references are not passed for fields
    that explicitly prevent them or require pydantic validation.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using plain-text secrets.
        **kwargs: Arguments to initialize this object.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            or an attribute which explicitly disallows secret references
            is passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}`. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure values with secrets "
                    "here: https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if secret_utils.is_clear_text_field(field):
            raise ValueError(
                f"Passing the `{key}` attribute as a secret reference is "
                "not allowed."
            )

        requires_validation = has_validators(
            pydantic_class=self.__class__, field_name=key
        )
        if requires_validation:
            raise ValueError(
                f"Passing the attribute `{key}` as a secret reference is "
                "not allowed as additional validation is required for "
                "this attribute."
            )

    super().__init__(**kwargs)

Functions

containerized_deployer

Base class for all ZenML deployers.

Classes

ContainerizedDeployer(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Bases: BaseDeployer, ABC

Base class for all containerized deployers.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        environment: Environment variables to set when running on this
            component.
        secrets: Secrets to set as environment variables when running on
            this component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.environment = environment or {}
    self.secrets = secrets or []
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None

Attributes

requirements: Set[str] property

Set of PyPI requirements for the deployer.

Returns:

Type	Description
Set[str]	A set of PyPI requirements for the deployer.

Functions

get_docker_builds(snapshot: PipelineSnapshotBase) -> List[BuildConfiguration]

Gets the Docker builds required for the component.

Parameters:

Name	Type	Description	Default
snapshot	PipelineSnapshotBase	The pipeline snapshot for which to get the builds.	required

Returns:

Type	Description
List[BuildConfiguration]	The required Docker builds.

Source code in src/zenml/deployers/containerized_deployer.py

def get_docker_builds(
    self, snapshot: "PipelineSnapshotBase"
) -> List["BuildConfiguration"]:
    """Gets the Docker builds required for the component.

    Args:
        snapshot: The pipeline snapshot for which to get the builds.

    Returns:
        The required Docker builds.
    """
    return [
        BuildConfiguration(
            key=DEPLOYER_DOCKER_IMAGE_KEY,
            settings=snapshot.pipeline_configuration.docker_settings,
        )
    ]

get_image(snapshot: PipelineSnapshotResponse) -> str staticmethod

Get the docker image used to deploy a pipeline snapshot.

Parameters:

Name	Type	Description	Default
snapshot	PipelineSnapshotResponse	The pipeline snapshot to get the image for.	required

Returns:

Type	Description
str	The docker image used to deploy the pipeline snapshot.

Raises:

Type	Description
RuntimeError	if the pipeline snapshot does not have a build or if the deployer image is not in the build.

Source code in src/zenml/deployers/containerized_deployer.py

@staticmethod
def get_image(snapshot: PipelineSnapshotResponse) -> str:
    """Get the docker image used to deploy a pipeline snapshot.

    Args:
        snapshot: The pipeline snapshot to get the image for.

    Returns:
        The docker image used to deploy the pipeline snapshot.

    Raises:
        RuntimeError: if the pipeline snapshot does not have a build or
            if the deployer image is not in the build.
    """
    if snapshot.build is None:
        raise RuntimeError("Pipeline snapshot does not have a build. ")
    if DEPLOYER_DOCKER_IMAGE_KEY not in snapshot.build.images:
        raise RuntimeError(
            "Pipeline snapshot build does not have a deployer image. "
        )
    return snapshot.build.images[DEPLOYER_DOCKER_IMAGE_KEY].image

Functions

docker

Implementation for the local Docker deployer.

Modules

docker_deployer

Implementation of the ZenML Docker deployer.

Classes

DockerDeployer(name: str, id: UUID, config: StackComponentConfig, flavor: str, type: StackComponentType, user: Optional[UUID], created: datetime, updated: datetime, environment: Optional[Dict[str, str]] = None, secrets: Optional[List[UUID]] = None, labels: Optional[Dict[str, Any]] = None, connector_requirements: Optional[ServiceConnectorRequirements] = None, connector: Optional[UUID] = None, connector_resource_id: Optional[str] = None, *args: Any, **kwargs: Any)

Bases: ContainerizedDeployer

Deployer responsible for deploying pipelines locally using Docker.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self,
    name: str,
    id: UUID,
    config: StackComponentConfig,
    flavor: str,
    type: StackComponentType,
    user: Optional[UUID],
    created: datetime,
    updated: datetime,
    environment: Optional[Dict[str, str]] = None,
    secrets: Optional[List[UUID]] = None,
    labels: Optional[Dict[str, Any]] = None,
    connector_requirements: Optional[ServiceConnectorRequirements] = None,
    connector: Optional[UUID] = None,
    connector_resource_id: Optional[str] = None,
    *args: Any,
    **kwargs: Any,
):
    """Initializes a StackComponent.

    Args:
        name: The name of the component.
        id: The unique ID of the component.
        config: The config of the component.
        flavor: The flavor of the component.
        type: The type of the component.
        user: The ID of the user who created the component.
        created: The creation time of the component.
        updated: The last update time of the component.
        environment: Environment variables to set when running on this
            component.
        secrets: Secrets to set as environment variables when running on
            this component.
        labels: The labels of the component.
        connector_requirements: The requirements for the connector.
        connector: The ID of a connector linked to the component.
        connector_resource_id: The custom resource ID to access through
            the connector.
        *args: Additional positional arguments.
        **kwargs: Additional keyword arguments.

    Raises:
        ValueError: If a secret reference is passed as name.
    """
    if secret_utils.is_secret_reference(name):
        raise ValueError(
            "Passing the `name` attribute of a stack component as a "
            "secret reference is not allowed."
        )

    self.id = id
    self.name = name
    self._config = config
    self.flavor = flavor
    self.type = type
    self.user = user
    self.created = created
    self.updated = updated
    self.labels = labels
    self.environment = environment or {}
    self.secrets = secrets or []
    self.connector_requirements = connector_requirements
    self.connector = connector
    self.connector_resource_id = connector_resource_id
    self._connector_instance: Optional[ServiceConnector] = None

Attributes

config: DockerDeployerConfig property

Returns the DockerDeployerConfig config.

Returns:

Type	Description
DockerDeployerConfig	The configuration.

docker_client: DockerClient property

Initialize and/or return the docker client.

Returns:

Type	Description
DockerClient	The docker client.

settings_class: Optional[Type[BaseSettings]] property

Settings class for the Docker deployer.

Returns:

Type	Description
Optional[Type[BaseSettings]]	The settings class.

validator: Optional[StackValidator] property

Ensures there is an image builder in the stack.

Returns:

Type	Description
Optional[StackValidator]	A StackValidator instance.

Functions

do_deprovision_deployment(deployment: DeploymentResponse, timeout: int) -> Optional[DeploymentOperationalState]

Deprovision a docker deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to deprovision.	required
timeout	int	The maximum time in seconds to wait for the pipeline deployment to be deprovisioned.	required

Returns:

Type	Description
Optional[DeploymentOperationalState]	The DeploymentOperationalState object representing the
Optional[DeploymentOperationalState]	operational state of the deleted deployment, or None if the
Optional[DeploymentOperationalState]	deletion is completed before the call returns.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.
DeploymentDeprovisionError	if the deployment deprovision fails.

Source code in src/zenml/deployers/docker/docker_deployer.py

def do_deprovision_deployment(
    self,
    deployment: DeploymentResponse,
    timeout: int,
) -> Optional[DeploymentOperationalState]:
    """Deprovision a docker deployment.

    Args:
        deployment: The deployment to deprovision.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be deprovisioned.

    Returns:
        The DeploymentOperationalState object representing the
        operational state of the deleted deployment, or None if the
        deletion is completed before the call returns.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
        DeploymentDeprovisionError: if the deployment
            deprovision fails.
    """
    container = self._get_container(deployment)
    if container is None:
        raise DeploymentNotFoundError(
            f"Docker container for deployment '{deployment.name}' "
            "not found"
        )

    try:
        container.stop(timeout=timeout)
        container.remove()
    except docker_errors.DockerException as e:
        raise DeploymentDeprovisionError(
            f"Docker container for deployment '{deployment.name}' "
            f"failed to delete: {e}"
        )

    return None

do_get_deployment_state(deployment: DeploymentResponse) -> DeploymentOperationalState

Get information about a docker deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to get information about.	required

Returns:

Type	Description
DeploymentOperationalState	The DeploymentOperationalState object representing the
DeploymentOperationalState	updated operational state of the deployment.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.

Source code in src/zenml/deployers/docker/docker_deployer.py

def do_get_deployment_state(
    self,
    deployment: DeploymentResponse,
) -> DeploymentOperationalState:
    """Get information about a docker deployment.

    Args:
        deployment: The deployment to get information about.

    Returns:
        The DeploymentOperationalState object representing the
        updated operational state of the deployment.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
    """
    container = self._get_container(deployment)
    if container is None:
        raise DeploymentNotFoundError(
            f"Docker container for deployment '{deployment.name}' "
            "not found"
        )

    return self._get_container_operational_state(container)

do_get_deployment_state_logs(deployment: DeploymentResponse, follow: bool = False, tail: Optional[int] = None) -> Generator[str, bool, None]

Get the logs of a Docker deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to get the logs of.	required
follow	bool	if True, the logs will be streamed as they are written	False
tail	Optional[int]	only retrieve the last NUM lines of log output.	None

Yields:

Type	Description
str	The logs of the deployment.

Raises:

Type	Description
DeploymentNotFoundError	if no deployment is found corresponding to the provided DeploymentResponse.
DeploymentLogsNotFoundError	if the deployment logs are not found.
DeployerError	if the deployment logs cannot be retrieved for any other reason or if an unexpected error occurs.

Source code in src/zenml/deployers/docker/docker_deployer.py

def do_get_deployment_state_logs(
    self,
    deployment: DeploymentResponse,
    follow: bool = False,
    tail: Optional[int] = None,
) -> Generator[str, bool, None]:
    """Get the logs of a Docker deployment.

    Args:
        deployment: The deployment to get the logs of.
        follow: if True, the logs will be streamed as they are written
        tail: only retrieve the last NUM lines of log output.

    Yields:
        The logs of the deployment.

    Raises:
        DeploymentNotFoundError: if no deployment is found
            corresponding to the provided DeploymentResponse.
        DeploymentLogsNotFoundError: if the deployment logs are not
            found.
        DeployerError: if the deployment logs cannot
            be retrieved for any other reason or if an unexpected error
            occurs.
    """
    container = self._get_container(deployment)
    if container is None:
        raise DeploymentNotFoundError(
            f"Docker container for deployment '{deployment.name}' "
            "not found"
        )

    try:
        log_kwargs: Dict[str, Any] = {
            "stdout": True,
            "stderr": True,
            "stream": follow,
            "follow": follow,
            "timestamps": True,
        }

        if tail is not None and tail > 0:
            log_kwargs["tail"] = tail

        log_stream = container.logs(**log_kwargs)

        if follow:
            for log_line in log_stream:
                if isinstance(log_line, bytes):
                    yield log_line.decode(
                        "utf-8", errors="replace"
                    ).rstrip()
                else:
                    yield str(log_line).rstrip()
        else:
            if isinstance(log_stream, bytes):
                log_text = log_stream.decode("utf-8", errors="replace")
                for line in log_text.splitlines():
                    yield line
            else:
                for log_line in log_stream:
                    if isinstance(log_line, bytes):
                        yield log_line.decode(
                            "utf-8", errors="replace"
                        ).rstrip()
                    else:
                        yield str(log_line).rstrip()

    except docker_errors.NotFound as e:
        raise DeploymentLogsNotFoundError(
            f"Logs for deployment '{deployment.name}' not found: {e}"
        )
    except docker_errors.APIError as e:
        raise DeployerError(
            f"Docker API error while retrieving logs for deployment "
            f"'{deployment.name}': {e}"
        )
    except docker_errors.DockerException as e:
        raise DeployerError(
            f"Docker error while retrieving logs for deployment "
            f"'{deployment.name}': {e}"
        )
    except Exception as e:
        raise DeployerError(
            f"Unexpected error while retrieving logs for deployment "
            f"'{deployment.name}': {e}"
        )

do_provision_deployment(deployment: DeploymentResponse, stack: Stack, environment: Dict[str, str], secrets: Dict[str, str], timeout: int) -> DeploymentOperationalState

Deploy a pipeline as a Docker container.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to run as a Docker container.	required
stack	Stack	The stack the pipeline will be deployed on.	required
environment	Dict[str, str]	A dictionary of environment variables to set on the deployment.	required
secrets	Dict[str, str]	A dictionary of secret environment variables to set on the deployment. These secret environment variables should not be exposed as regular environment variables on the deployer.	required
timeout	int	The maximum time in seconds to wait for the pipeline deployment to be provisioned.	required

Returns:

Type	Description
DeploymentOperationalState	The DeploymentOperationalState object representing the
DeploymentOperationalState	operational state of the provisioned deployment.

Raises:

Type	Description
DeploymentProvisionError	if provisioning the deployment fails.

Source code in src/zenml/deployers/docker/docker_deployer.py

def do_provision_deployment(
    self,
    deployment: DeploymentResponse,
    stack: "Stack",
    environment: Dict[str, str],
    secrets: Dict[str, str],
    timeout: int,
) -> DeploymentOperationalState:
    """Deploy a pipeline as a Docker container.

    Args:
        deployment: The deployment to run as a Docker container.
        stack: The stack the pipeline will be deployed on.
        environment: A dictionary of environment variables to set on the
            deployment.
        secrets: A dictionary of secret environment variables to set
            on the deployment. These secret environment variables
            should not be exposed as regular environment variables on the
            deployer.
        timeout: The maximum time in seconds to wait for the pipeline
            deployment to be provisioned.

    Returns:
        The DeploymentOperationalState object representing the
        operational state of the provisioned deployment.

    Raises:
        DeploymentProvisionError: if provisioning the deployment
            fails.
    """
    assert deployment.snapshot, "Pipeline snapshot not found"
    snapshot = deployment.snapshot

    # Currently, there is no safe way to pass secrets to a docker
    # container, so we simply merge them into the environment variables.
    environment.update(secrets)

    settings = cast(
        DockerDeployerSettings,
        self.get_settings(snapshot),
    )

    existing_metadata = DockerDeploymentMetadata.from_deployment(
        deployment
    )

    entrypoint = DeploymentEntrypointConfiguration.get_entrypoint_command()

    entrypoint_kwargs = {
        DEPLOYMENT_ID_OPTION: deployment.id,
        PORT_OPTION: 8000,
    }
    if deployment.auth_key:
        entrypoint_kwargs[AUTH_KEY_OPTION] = deployment.auth_key

    arguments = DeploymentEntrypointConfiguration.get_entrypoint_arguments(
        **entrypoint_kwargs
    )

    # Add the local stores path as a volume mount
    stack.check_local_paths()
    local_stores_path = GlobalConfiguration().local_stores_path
    volumes = {
        local_stores_path: {
            "bind": local_stores_path,
            "mode": "rw",
        }
    }
    environment[ENV_ZENML_LOCAL_STORES_PATH] = local_stores_path

    # check if a container already exists for the deployment
    container = self._get_container(deployment)

    if container:
        # the container exists, check if it is running
        if container.status == "running":
            logger.debug(
                f"Container for deployment '{deployment.name}' is "
                "already running",
            )
            container.stop(timeout=timeout)

        # the container is stopped or in an error state, remove it
        logger.debug(
            f"Removing previous container for deployment "
            f"'{deployment.name}'",
        )
        container.remove(force=True)

    logger.debug(
        f"Starting container for deployment '{deployment.name}'..."
    )

    image = self.get_image(deployment.snapshot)

    try:
        self.docker_client.images.get(image)
    except docker_errors.ImageNotFound:
        logger.debug(
            f"Pulling container image '{image}' for deployment "
            f"'{deployment.name}'...",
        )
        self.docker_client.images.pull(image)

    preferred_ports: List[int] = []
    if settings.port:
        preferred_ports.append(settings.port)
    if existing_metadata.port:
        preferred_ports.append(existing_metadata.port)
    port = lookup_preferred_or_free_port(
        preferred_ports=preferred_ports,
        allocate_port_if_busy=settings.allocate_port_if_busy,
        range=settings.port_range,
    )
    ports: Dict[str, Optional[int]] = {"8000/tcp": port}

    uid_args: Dict[str, Any] = {}
    if sys.platform == "win32":
        # File permissions are not checked on Windows. This if clause
        # prevents mypy from complaining about unused 'type: ignore'
        # statements
        pass
    else:
        # Run the container in the context of the local UID/GID
        # to ensure that the local database can be shared
        # with the container.
        logger.debug(
            "Setting UID and GID to local user/group in container."
        )
        uid_args = dict(
            user=os.getuid(),
            group_add=[os.getgid()],
        )

    run_args = copy.deepcopy(settings.run_args)
    docker_environment = run_args.pop("environment", {})
    docker_environment.update(environment)

    docker_volumes = run_args.pop("volumes", {})
    docker_volumes.update(volumes)

    extra_hosts = run_args.pop("extra_hosts", {})
    extra_hosts["host.docker.internal"] = "host-gateway"

    run_args.update(uid_args)

    try:
        container = self.docker_client.containers.run(
            image=image,
            name=self._get_container_id(deployment),
            entrypoint=entrypoint,
            command=arguments,
            detach=True,
            volumes=docker_volumes,
            environment=docker_environment,
            remove=False,
            auto_remove=False,
            ports=ports,
            labels={
                "zenml-deployment-id": str(deployment.id),
                "zenml-deployment-name": deployment.name,
                "zenml-deployer-name": str(self.name),
                "zenml-deployer-id": str(self.id),
                "managed-by": "zenml",
            },
            extra_hosts=extra_hosts,
            **run_args,
        )

        logger.debug(
            f"Docker container for deployment '{deployment.name}' "
            f"started with ID {self._get_container_id(deployment)}",
        )

    except docker_errors.DockerException as e:
        raise DeploymentProvisionError(
            f"Docker container for deployment '{deployment.name}' "
            f"failed to start: {e}"
        )

    return self._get_container_operational_state(container)

DockerDeployerConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseDeployerConfig, DockerDeployerSettings

Docker deployer config.

Source code in src/zenml/stack/stack_component.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references don't clash with pydantic validation.

    StackComponents allow the specification of all their string attributes
    using secret references of the form `{{secret_name.key}}`. This however
    is only possible when the stack component does not perform any explicit
    validation of this attribute using pydantic validators. If this were
    the case, the validation would run on the secret reference and would
    fail or in the worst case, modify the secret reference and lead to
    unexpected behavior. This method ensures that no attributes that require
    custom pydantic validation are set as secret references.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using
            plain-text secrets.
        **kwargs: Arguments to initialize this stack component.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            is passed as a secret reference, or if the `name` attribute
            was passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}` for a `{self.__class__.__name__}` "
                    "stack component. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure your stack "
                    "components with secrets here: "
                    "https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if pydantic_utils.has_validators(
            pydantic_class=self.__class__, field_name=key
        ):
            raise ValueError(
                f"Passing the stack component attribute `{key}` as a "
                "secret reference is not allowed as additional validation "
                "is required for this attribute."
            )

    super().__init__(**kwargs)

Attributes

is_local: bool property

Checks if this stack component is running locally.

Returns:

Type	Description
bool	True if this config is for a local component, False otherwise.

DockerDeployerFlavor

Bases: BaseDeployerFlavor

Flavor for the Docker deployer.

Attributes

config_class: Type[BaseDeployerConfig] property

Config class for the base deployer flavor.

Returns:

Type	Description
Type[BaseDeployerConfig]	The config class.

docs_url: Optional[str] property

A url to point at docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor docs url.

implementation_class: Type[DockerDeployer] property

Implementation class for this flavor.

Returns:

Type	Description
Type[DockerDeployer]	Implementation class for this flavor.

logo_url: str property

A url to represent the flavor in the dashboard.

Returns:

Type	Description
str	The flavor logo.

name: str property

Name of the deployer flavor.

Returns:

Type	Description
str	Name of the deployer flavor.

sdk_docs_url: Optional[str] property

A url to point at SDK docs explaining this flavor.

Returns:

Type	Description
Optional[str]	A flavor SDK docs url.

DockerDeployerSettings(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseDeployerSettings

Docker deployer settings.

Attributes:

Name	Type	Description
port	Optional[int]	The port to expose the deployment on.
allocate_port_if_busy	bool	If True, allocate a free port if the configured port is busy.
port_range	Tuple[int, int]	The range of ports to search for a free port.
run_args	Dict[str, Any]	Arguments to pass to the docker run call. (See https://docker-py.readthedocs.io/en/stable/containers.html for a list of what can be passed.)

Source code in src/zenml/config/secret_reference_mixin.py

def __init__(
    self, warn_about_plain_text_secrets: bool = False, **kwargs: Any
) -> None:
    """Ensures that secret references are only passed for valid fields.

    This method ensures that secret references are not passed for fields
    that explicitly prevent them or require pydantic validation.

    Args:
        warn_about_plain_text_secrets: If true, then warns about using plain-text secrets.
        **kwargs: Arguments to initialize this object.

    Raises:
        ValueError: If an attribute that requires custom pydantic validation
            or an attribute which explicitly disallows secret references
            is passed as a secret reference.
    """
    for key, value in kwargs.items():
        try:
            field = self.__class__.model_fields[key]
        except KeyError:
            # Value for a private attribute or non-existing field, this
            # will fail during the upcoming pydantic validation
            continue

        if value is None:
            continue

        if not secret_utils.is_secret_reference(value):
            if (
                secret_utils.is_secret_field(field)
                and warn_about_plain_text_secrets
            ):
                logger.warning(
                    "You specified a plain-text value for the sensitive "
                    f"attribute `{key}`. This is currently only a warning, "
                    "but future versions of ZenML will require you to pass "
                    "in sensitive information as secrets. Check out the "
                    "documentation on how to configure values with secrets "
                    "here: https://docs.zenml.io/deploying-zenml/deploying-zenml/secret-management"
                )
            continue

        if secret_utils.is_clear_text_field(field):
            raise ValueError(
                f"Passing the `{key}` attribute as a secret reference is "
                "not allowed."
            )

        requires_validation = has_validators(
            pydantic_class=self.__class__, field_name=key
        )
        if requires_validation:
            raise ValueError(
                f"Passing the attribute `{key}` as a secret reference is "
                "not allowed as additional validation is required for "
                "this attribute."
            )

    super().__init__(**kwargs)

DockerDeploymentMetadata

Bases: BaseModel

Metadata for a Docker deployment.

Functions

from_container(container: Container) -> DockerDeploymentMetadata classmethod

Create a DockerDeploymentMetadata from a docker container.

Parameters:

Name	Type	Description	Default
container	Container	The docker container to get the metadata for.	required

Returns:

Type	Description
DockerDeploymentMetadata	The metadata for the docker container.

Source code in src/zenml/deployers/docker/docker_deployer.py

@classmethod
def from_container(
    cls, container: Container
) -> "DockerDeploymentMetadata":
    """Create a DockerDeploymentMetadata from a docker container.

    Args:
        container: The docker container to get the metadata for.

    Returns:
        The metadata for the docker container.
    """
    image = container.image
    if image is not None:
        image_url = image.attrs["RepoTags"][0]
        image_id = image.attrs["Id"]
    else:
        image_url = None
        image_id = None
    if container.ports:
        ports = list(container.ports.values())
        if len(ports) > 0:
            port = int(ports[0][0]["HostPort"])
        else:
            port = None
    else:
        port = None
    return cls(
        port=port,
        container_id=container.id,
        container_name=container.name,
        container_image_uri=image_url,
        container_image_id=image_id,
        container_status=container.status,
    )

from_deployment(deployment: DeploymentResponse) -> DockerDeploymentMetadata classmethod

Create a DockerDeploymentMetadata from a deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment to get the metadata for.	required

Returns:

Type	Description
DockerDeploymentMetadata	The metadata for the deployment.

Source code in src/zenml/deployers/docker/docker_deployer.py

@classmethod
def from_deployment(
    cls, deployment: DeploymentResponse
) -> "DockerDeploymentMetadata":
    """Create a DockerDeploymentMetadata from a deployment.

    Args:
        deployment: The deployment to get the metadata for.

    Returns:
        The metadata for the deployment.
    """
    return cls.model_validate(deployment.deployment_metadata)

Functions Modules

exceptions

Base class for all ZenML deployers.

Classes

DeployerError

Bases: Exception

Base class for deployer errors.

DeploymentAlreadyExistsError(message: Optional[str] = None, url: Optional[str] = None)

Bases: EntityExistsError, DeployerError

Error raised when a deployment already exists.

Source code in src/zenml/exceptions.py

def __init__(
    self,
    message: Optional[str] = None,
    url: Optional[str] = None,
):
    """The BaseException used to format messages displayed to the user.

    Args:
        message: Message with details of exception. This message
                 will be appended with another message directing user to
                 `url` for more information. If `None`, then default
                 Exception behavior is used.
        url: URL to point to in exception message. If `None`, then no url
             is appended.
    """
    if message and url:
        message += f" For more information, visit {url}."
    super().__init__(message)

DeploymentDeployerMismatchError

Bases: DeployerError

Error raised when a deployment is not managed by this deployer.

DeploymentDeprovisionError

Bases: DeployerError

Error raised when a deployment deprovisioning fails.

DeploymentHTTPError

Bases: DeployerError

Error raised when an HTTP request to a deployment fails.

DeploymentInvalidParametersError

Bases: DeployerError

Error raised when the parameters for a deployment are invalid.

DeploymentLogsNotFoundError

Bases: KeyError, DeployerError

Error raised when pipeline logs are not found.

DeploymentNotFoundError

Bases: KeyError, DeployerError

Error raised when a deployment is not found.

DeploymentProvisionError

Bases: DeployerError

Error raised when a deployment provisioning fails.

DeploymentSnapshotMismatchError

Bases: DeployerError

Error raised when a deployment snapshot does not match the current deployer.

DeploymentTimeoutError

Bases: DeployerError

Error raised when a deployment provisioning or deprovisioning times out.

Functions

utils

ZenML deployers utilities.

Classes

Functions

get_deployment_input_schema(deployment: DeploymentResponse) -> Dict[str, Any]

Get the schema for a deployment's input parameters.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment for which to get the schema.	required

Returns:

Type	Description
Dict[str, Any]	The schema for the deployment's input parameters.

Raises:

Type	Description
RuntimeError	If the deployment has no associated input schema.

Source code in src/zenml/deployers/utils.py

def get_deployment_input_schema(
    deployment: DeploymentResponse,
) -> Dict[str, Any]:
    """Get the schema for a deployment's input parameters.

    Args:
        deployment: The deployment for which to get the schema.

    Returns:
        The schema for the deployment's input parameters.

    Raises:
        RuntimeError: If the deployment has no associated input schema.
    """
    if (
        deployment.snapshot
        and deployment.snapshot.pipeline_spec
        and deployment.snapshot.pipeline_spec.input_schema
    ):
        return deployment.snapshot.pipeline_spec.input_schema

    raise RuntimeError(
        f"Deployment {deployment.name} has no associated input schema."
    )

get_deployment_invocation_example(deployment: DeploymentResponse) -> Dict[str, Any]

Generate an example invocation command for a deployment.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment for which to generate an example invocation.	required

Returns:

Type	Description
Dict[str, Any]	A dictionary containing the example invocation parameters.

Source code in src/zenml/deployers/utils.py

def get_deployment_invocation_example(
    deployment: DeploymentResponse,
) -> Dict[str, Any]:
    """Generate an example invocation command for a deployment.

    Args:
        deployment: The deployment for which to generate an example invocation.

    Returns:
        A dictionary containing the example invocation parameters.
    """
    parameters_schema = get_deployment_input_schema(deployment)

    properties = parameters_schema.get("properties", {})

    if not properties:
        return {}

    parameters = {}

    for attr_name, attr_schema in properties.items():
        parameters[attr_name] = "<value>"
        if not isinstance(attr_schema, dict):
            continue

        default_value = None

        if "default" in attr_schema:
            default_value = attr_schema["default"]
        elif "const" in attr_schema:
            default_value = attr_schema["const"]

        parameters[attr_name] = default_value or "<value>"

    return parameters

get_deployment_output_schema(deployment: DeploymentResponse) -> Dict[str, Any]

Get the schema for a deployment's output parameters.

Parameters:

Name	Type	Description	Default
deployment	DeploymentResponse	The deployment for which to get the schema.	required

Returns:

Type	Description
Dict[str, Any]	The schema for the deployment's output parameters.

Raises:

Type	Description
RuntimeError	If the deployment has no associated output schema.

Source code in src/zenml/deployers/utils.py

def get_deployment_output_schema(
    deployment: DeploymentResponse,
) -> Dict[str, Any]:
    """Get the schema for a deployment's output parameters.

    Args:
        deployment: The deployment for which to get the schema.

    Returns:
        The schema for the deployment's output parameters.

    Raises:
        RuntimeError: If the deployment has no associated output schema.
    """
    if (
        deployment.snapshot
        and deployment.snapshot.pipeline_spec
        and deployment.snapshot.pipeline_spec.output_schema
    ):
        return deployment.snapshot.pipeline_spec.output_schema

    raise RuntimeError(
        f"Deployment {deployment.name} has no associated output schema."
    )

invoke_deployment(deployment_name_or_id: Union[str, UUID], project: Optional[UUID] = None, timeout: int = 300, **kwargs: Any) -> Any

Call a deployment and return the result.

Parameters:

Name	Type	Description	Default
deployment_name_or_id	Union[str, UUID]	The name or ID of the deployment to call.	required
project	Optional[UUID]	The project ID of the deployment to call.	None
timeout	int	The timeout for the HTTP request to the deployment.	300
**kwargs	Any	Keyword arguments to pass to the deployment.	{}

Returns:

Type	Description
Any	The response from the deployment, parsed as JSON if possible,
Any	otherwise returned as text.

Raises:

Type	Description
DeploymentNotFoundError	If the deployment is not found.
DeploymentProvisionError	If the deployment is not running or has no URL.
DeploymentHTTPError	If the HTTP request to the endpoint fails.

Source code in src/zenml/deployers/utils.py

def invoke_deployment(
    deployment_name_or_id: Union[str, UUID],
    project: Optional[UUID] = None,
    timeout: int = 300,  # 5 minute timeout
    **kwargs: Any,
) -> Any:
    """Call a deployment and return the result.

    Args:
        deployment_name_or_id: The name or ID of the deployment to call.
        project: The project ID of the deployment to call.
        timeout: The timeout for the HTTP request to the deployment.
        **kwargs: Keyword arguments to pass to the deployment.

    Returns:
        The response from the deployment, parsed as JSON if possible,
        otherwise returned as text.

    Raises:
        DeploymentNotFoundError: If the deployment is not found.
        DeploymentProvisionError: If the deployment is not running
            or has no URL.
        DeploymentHTTPError: If the HTTP request to the endpoint fails.
    """
    client = Client()
    try:
        deployment = client.get_deployment(
            deployment_name_or_id, project=project
        )
    except KeyError:
        raise DeploymentNotFoundError(
            f"Deployment with name or ID '{deployment_name_or_id}' not found"
        )

    if deployment.status != DeploymentStatus.RUNNING:
        raise DeploymentProvisionError(
            f"Deployment {deployment_name_or_id} is not running. Please "
            "refresh or re-deploy the deployment or check its logs for "
            "more details."
        )

    if not deployment.url:
        raise DeploymentProvisionError(
            f"Deployment {deployment_name_or_id} has no URL. Please "
            "refresh the deployment or check its logs for more "
            "details."
        )

    input_schema = None
    if deployment.snapshot and deployment.snapshot.pipeline_spec:
        input_schema = deployment.snapshot.pipeline_spec.input_schema

    if input_schema:
        # Resolve the references in the schema first, otherwise we won't be able
        # to access the data types for object-typed parameters.
        input_schema = jsonref.replace_refs(input_schema)
        assert isinstance(input_schema, dict)

        properties = input_schema.get("properties", {})

        # Some kwargs having one of the collection data types (list, dict) in
        # the schema may be supplied as a JSON string. We need to unpack
        # them before we construct the final JSON payload.
        #
        # We ignore all errors here because they will be better handled by the
        # deployment itself server side.
        for key in kwargs.keys():
            if key not in properties:
                continue
            value = kwargs[key]
            if not isinstance(value, str):
                continue
            attr_schema = properties[key]
            try:
                if attr_schema.get("type") == "object":
                    value = json.loads(value)
                    if isinstance(value, dict):
                        kwargs[key] = value
                elif attr_schema.get("type") == "array":
                    value = json.loads(value)
                    if isinstance(value, list):
                        kwargs[key] = value
            except (json.JSONDecodeError, ValueError):
                pass

    # Serialize kwargs to JSON
    params = dict(parameters=kwargs)
    try:
        payload = json.dumps(params, default=pydantic_encoder)
    except (TypeError, ValueError) as e:
        raise DeploymentHTTPError(
            f"Failed to serialize request data to JSON: {e}"
        )

    # Construct the invoke endpoint URL
    invoke_url = deployment.url.rstrip("/") + "/invoke"

    # Prepare headers
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
    }

    # Add authorization header if auth_key is present
    if deployment.auth_key:
        headers["Authorization"] = f"Bearer {deployment.auth_key}"

    try:
        step_context = get_step_context()
    except RuntimeError:
        step_context = None

    if step_context:
        # Include these so that the deployment can identify the step
        # and pipeline run that called it, if called from a step.
        headers["ZenML-Step-Name"] = step_context.step_name
        headers["ZenML-Pipeline-Name"] = step_context.pipeline.name
        headers["ZenML-Pipeline-Run-ID"] = str(step_context.pipeline_run.id)
        headers["ZenML-Pipeline-Run-Name"] = step_context.pipeline_run.name

    # Make the HTTP request
    try:
        response = requests.post(
            invoke_url,
            data=payload,
            headers=headers,
            timeout=timeout,
        )
        response.raise_for_status()

        # Try to parse JSON response, fallback to text if not JSON
        try:
            return response.json()
        except ValueError:
            return response.text

    except requests.exceptions.HTTPError as e:
        raise DeploymentHTTPError(
            f"HTTP {e.response.status_code} error calling deployment "
            f"{deployment_name_or_id}: {e.response.text}"
        )
    except requests.exceptions.ConnectionError as e:
        raise DeploymentHTTPError(
            f"Failed to connect to deployment {deployment_name_or_id}: {e}"
        )
    except requests.exceptions.Timeout as e:
        raise DeploymentHTTPError(
            f"Timeout calling deployment {deployment_name_or_id}: {e}"
        )
    except requests.exceptions.RequestException as e:
        raise DeploymentHTTPError(
            f"Request failed for deployment {deployment_name_or_id}: {e}"
        )