Skip to content

Sandboxes

zenml.sandboxes

Sandbox stack components for isolated code execution.

Attributes

DOCKER_SANDBOX_FLAVOR = 'docker' module-attribute

LOCAL_SANDBOX_FLAVOR = 'local' module-attribute

__all__ = ['DOCKER_SANDBOX_FLAVOR', 'LOCAL_SANDBOX_FLAVOR', 'BaseSandbox', 'BaseSandboxConfig', 'BaseSandboxFlavor', 'BaseSandboxSettings', 'ContainerizedSandboxSettings', 'DockerSandbox', 'DockerSandboxConfig', 'DockerSandboxFlavor', 'DockerSandboxPullPolicy', 'DockerSandboxSettings', 'SandboxSnapshot', 'LocalSandbox', 'LocalSandboxConfig', 'LocalSandboxFlavor', 'LocalSandboxSettings', 'SandboxExecError', 'SandboxOutput', 'SandboxProcess', 'SandboxSession', 'SandboxSessionClosedError'] module-attribute

Classes

BaseSandbox(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 sandbox components.

Source code in src/zenml/stack/stack_component.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
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: BaseSandboxConfig property

The sandbox component configuration.

Returns:

Type Description
BaseSandboxConfig

The sandbox component configuration.
settings_class: Optional[Type[BaseSettings]] property

The sandbox settings class.

Returns:

Type Description
Optional[Type[BaseSettings]]

The sandbox settings class.
Methods:
attach(session_id: str) -> SandboxSession

Attach to a running sandbox session.

Parameters:

Name Type Description Default
session_id str

The ID of the running sandbox session.

 required

Raises:

Type Description
NotImplementedError

If the sandbox does not support attaching to a running sandbox session.

Returns:

Type Description
SandboxSession

Sandbox session.
Source code in src/zenml/sandboxes/base.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def attach(self, session_id: str) -> SandboxSession:
    """Attach to a running sandbox session.

    Args:
        session_id: The ID of the running sandbox session.

    Raises:
        NotImplementedError: If the sandbox does not support attaching to a
            running sandbox session.

    Returns:
        Sandbox session.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support attaching to a running "
        "sandbox session."
    )
create_session(settings: Optional[BaseSandboxSettings] = None, destroy_on_exit: bool = False) -> SandboxSession abstractmethod

Create a fresh sandbox session.

Parameters:

Name Type Description Default
settings Optional[BaseSandboxSettings]

The sandbox settings.

 None
destroy_on_exit bool

Whether to destroy the sandbox session when the session context manager exits.

 False

Returns:

Type Description
SandboxSession

A new sandbox session.
Source code in src/zenml/sandboxes/base.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@abstractmethod
def create_session(
    self,
    settings: Optional[BaseSandboxSettings] = None,
    destroy_on_exit: bool = False,
) -> SandboxSession:
    """Create a fresh sandbox session.

    Args:
        settings: The sandbox settings.
        destroy_on_exit: Whether to destroy the sandbox session when the
            session context manager exits.

    Returns:
        A new sandbox session.
    """
resolve_settings(override: Optional[BaseSandboxSettings] = None) -> BaseSandboxSettings

Resolve the sandbox settings.

Parameters:

Name Type Description Default
override Optional[BaseSandboxSettings]

Per-call settings override. If provided these will override values in the component settings/config.

 None

Returns:

Type Description
BaseSandboxSettings

The resolved settings.
Source code in src/zenml/sandboxes/base.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def resolve_settings(
    self, override: Optional[BaseSandboxSettings] = None
) -> BaseSandboxSettings:
    """Resolve the sandbox settings.

    Args:
        override: Per-call settings override. If provided these will
            override values in the component settings/config.

    Returns:
        The resolved settings.
    """
    from zenml.steps.step_context import StepContext

    if step_context := StepContext.get():
        settings_dict = self.get_settings(
            step_context.step_run
        ).model_dump(exclude_unset=True)
    else:
        settings_dict = self.config.model_dump(exclude_unset=True)

    if override is not None:
        settings_dict.update(override.model_dump(exclude_unset=True))

    assert self.settings_class is not None
    return cast(
        BaseSandboxSettings,
        self.settings_class.model_validate(settings_dict),
    )
restore(snapshot: SandboxSnapshot) -> SandboxSession

Restore a sandbox session from a snapshot.

Parameters:

Name Type Description Default
snapshot SandboxSnapshot

The snapshot to restore from.

 required

Raises:

Type Description
NotImplementedError

If the sandbox does not support restoring from a snapshot.

Returns:

Type Description
SandboxSession

Sandbox session.
Source code in src/zenml/sandboxes/base.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def restore(self, snapshot: SandboxSnapshot) -> SandboxSession:
    """Restore a sandbox session from a snapshot.

    Args:
        snapshot: The snapshot to restore from.

    Raises:
        NotImplementedError: If the sandbox does not support restoring from
            a snapshot.

    Returns:
        Sandbox session.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support restoring from a snapshot."
    )

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

Bases: StackComponentConfig

Sandbox configuration.

Source code in src/zenml/stack/stack_component.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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

Whether this sandbox component is running locally.

Returns:

Type Description
bool

Whether this sandbox component is running locally.

BaseSandboxFlavor

Bases: Flavor

Base sandbox flavor.

Attributes
config_class: Type[BaseSandboxConfig] property

Configuration class.

Returns:

Type Description
Type[BaseSandboxConfig]

The configuration class.
implementation_class: Type[BaseSandbox] abstractmethod property

Implementation class.

Returns:

Type Description
Type[BaseSandbox]

The implementation class.
type: StackComponentType property

Stack component type.

Returns:

Type Description
StackComponentType

StackComponentType.SANDBOX.

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

Bases: BaseSettings

Sandbox settings.

Source code in src/zenml/config/secret_reference_mixin.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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)

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

Bases: BaseSandboxSettings

Containerized sandbox settings.

Source code in src/zenml/config/secret_reference_mixin.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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)

DockerSandbox(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: BaseSandbox

Docker daemon-backed sandbox.

Source code in src/zenml/stack/stack_component.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
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: DockerSandboxConfig property

Docker sandbox configuration.

Returns:

Type Description
DockerSandboxConfig

The Docker sandbox configuration.
docker_client: DockerClient property

Initialize and/or return the docker client.

Returns:

Type Description
DockerClient

The docker client.
docker_engine: DockerContainerEngine property

Initialize and/or return the docker engine.

Returns:

Type Description
DockerContainerEngine

The docker engine.
settings_class: Optional[Type[BaseSandboxSettings]] property

Settings class.

Returns:

Type Description
Optional[Type[BaseSandboxSettings]]

DockerSandboxSettings.
Methods:
attach(session_id: str) -> SandboxSession

Attach to a running Docker sandbox session.

Parameters:

Name Type Description Default
session_id str

The ID of the running sandbox session.

 required

Raises:

Type Description
KeyError

If no running session with the given ID exists.

Returns:

Type Description
SandboxSession

Sandbox session.
Source code in src/zenml/sandboxes/docker_sandbox.py 
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
def attach(self, session_id: str) -> SandboxSession:
    """Attach to a running Docker sandbox session.

    Args:
        session_id: The ID of the running sandbox session.

    Raises:
        KeyError: If no running session with the given ID exists.

    Returns:
        Sandbox session.
    """
    containers = self.docker_client.containers.list(
        filters={"label": f"{_SESSION_ID_LABEL}={session_id}"}
    )
    if not containers:
        raise KeyError(
            f"No running Docker sandbox session `{session_id}` found."
        )

    settings = cast(DockerSandboxSettings, self.resolve_settings())
    return DockerSandboxSession(
        container=containers[0],
        workdir=settings.workdir,
        env=self._resolve_session_environment(settings),
        parent=self,
        id=session_id,
    )
create_session(settings: Optional[BaseSandboxSettings] = None, destroy_on_exit: bool = False) -> SandboxSession

Create a Docker sandbox session backed by a new container.

Parameters:

Name Type Description Default
settings Optional[BaseSandboxSettings]

Optional settings overrides.

 None
destroy_on_exit bool

Whether to destroy the sandbox session when the session context manager exits.

 False

Returns:

Type Description
SandboxSession

A Docker sandbox session.
Source code in src/zenml/sandboxes/docker_sandbox.py 
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
def create_session(
    self,
    settings: Optional[BaseSandboxSettings] = None,
    destroy_on_exit: bool = False,
) -> SandboxSession:
    """Create a Docker sandbox session backed by a new container.

    Args:
        settings: Optional settings overrides.
        destroy_on_exit: Whether to destroy the sandbox session when the
            session context manager exits.

    Returns:
        A Docker sandbox session.
    """
    settings = cast(DockerSandboxSettings, self.resolve_settings(settings))
    self._ensure_image(settings)

    session_id = f"docker-{uuid4().hex[:12]}"

    run_args = copy.deepcopy(settings.run_args)
    labels = run_args.pop("labels", {})
    labels[_SESSION_ID_LABEL] = session_id
    if settings.cpu_limit is not None:
        run_args["nano_cpus"] = int(settings.cpu_limit * 1e9)
    if settings.memory_limit is not None:
        run_args["mem_limit"] = settings.memory_limit

    container = self.docker_client.containers.run(
        image=settings.image,
        command=["sleep", "infinity"],
        entrypoint=[],
        detach=True,
        name=f"zenml-sandbox-{session_id}",
        labels=labels,
        working_dir=settings.workdir,
        **run_args,
    )

    return DockerSandboxSession(
        container=container,
        workdir=settings.workdir,
        env=self._resolve_session_environment(settings),
        parent=self,
        destroy_on_exit=destroy_on_exit,
        id=session_id,
    )
restore(snapshot: SandboxSnapshot) -> SandboxSession

Restore a sandbox session from a snapshot.

Parameters:

Name Type Description Default
snapshot SandboxSnapshot

The snapshot to restore from.

 required

Returns:

Type Description
SandboxSession

A new session started from the snapshot image.
Source code in src/zenml/sandboxes/docker_sandbox.py 
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
def restore(self, snapshot: SandboxSnapshot) -> SandboxSession:
    """Restore a sandbox session from a snapshot.

    Args:
        snapshot: The snapshot to restore from.

    Returns:
        A new session started from the snapshot image.
    """
    self._validate_snapshot(snapshot)
    # Snapshot images are stored locally, so they must never be pulled.
    return self.create_session(
        settings=DockerSandboxSettings(
            image=snapshot.ref,
            pull_policy=DockerSandboxPullPolicy.NEVER,
        )
    )

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

Bases: BaseSandboxConfig, DockerSandboxSettings

Docker sandbox configuration.

Source code in src/zenml/stack/stack_component.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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

Whether this sandbox component is running locally.

Returns:

Type Description
bool

Whether this sandbox component is running locally.

DockerSandboxFlavor

Bases: BaseSandboxFlavor

Docker sandbox flavor.

Attributes
config_class: Type[DockerSandboxConfig] property

Config class.

Returns:

Type Description
Type[DockerSandboxConfig]

DockerSandboxConfig.
docs_url: Optional[str] property

Docs URL.

Returns:

Type Description
Optional[str]

The docs URL.
implementation_class: Type[DockerSandbox] property

Implementation class.

Returns:

Type Description
Type[DockerSandbox]

DockerSandbox.
logo_url: str property

Dashboard logo URL.

Returns:

Type Description
str

The flavor logo URL.
name: str property

Flavor name.

Returns:

Type Description
str

The flavor name.
sdk_docs_url: Optional[str] property

SDK docs URL.

Returns:

Type Description
Optional[str]

The flavor SDK docs URL.

DockerSandboxPullPolicy

Bases: StrEnum

Docker sandbox image pull policy.

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

Bases: ContainerizedSandboxSettings

Docker sandbox settings.

Source code in src/zenml/config/secret_reference_mixin.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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)

LocalSandbox(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: BaseSandbox

Local subprocess-based Sandbox.

Warning: This class does NOT provide any isolation and is not suitable for running untrusted code.

Source code in src/zenml/stack/stack_component.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
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: LocalSandboxConfig property

Local sandbox configuration.

Returns:

Type Description
LocalSandboxConfig

The local sandbox configuration.
settings_class: Optional[Type[BaseSandboxSettings]] property

Settings class.

Returns:

Type Description
Optional[Type[BaseSandboxSettings]]

LocalSandboxSettings.
Methods:
create_session(settings: Optional[BaseSandboxSettings] = None, destroy_on_exit: bool = False) -> SandboxSession

Create a local sandbox session in a clean working directory.

Parameters:

Name Type Description Default
settings Optional[BaseSandboxSettings]

Optional settings overrides.

 None
destroy_on_exit bool

Whether to destroy the sandbox session when the session context manager exits.

 False

Returns:

Type Description
SandboxSession

A local sandbox session.
Source code in src/zenml/sandboxes/local_sandbox.py 
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
def create_session(
    self,
    settings: Optional[BaseSandboxSettings] = None,
    destroy_on_exit: bool = False,
) -> SandboxSession:
    """Create a local sandbox session in a clean working directory.

    Args:
        settings: Optional settings overrides.
        destroy_on_exit: Whether to destroy the sandbox session when the
            session context manager exits.

    Returns:
        A local sandbox session.
    """
    logger.warning(
        "The local sandbox does not provide any isolation. Any code "
        "executed in this sandbox will have full access to the local "
        "filesystem and network."
    )

    settings = cast(LocalSandboxSettings, self.resolve_settings(settings))
    env = self._resolve_session_environment(settings)
    workdir = tempfile.mkdtemp(prefix="zenml-local-sandbox-")

    return LocalSandboxSession(
        workdir=workdir,
        env=env,
        parent=self,
        destroy_on_exit=destroy_on_exit,
    )

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

Bases: BaseSandboxConfig, LocalSandboxSettings

Local sandbox configuration.

Source code in src/zenml/stack/stack_component.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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)

LocalSandboxFlavor

Bases: BaseSandboxFlavor

Local subprocess sandbox flavor.

Attributes
config_class: Type[LocalSandboxConfig] property

Config class.

Returns:

Type Description
Type[LocalSandboxConfig]

LocalSandboxConfig.
docs_url: Optional[str] property

Docs URL.

Returns:

Type Description
Optional[str]

The docs URL.
implementation_class: Type[LocalSandbox] property

Implementation class.

Returns:

Type Description
Type[LocalSandbox]

LocalSandbox.
logo_url: str property

Dashboard logo URL.

Returns:

Type Description
str

The flavor logo URL.
name: str property

Flavor name.

Returns:

Type Description
str

The flavor name.
sdk_docs_url: Optional[str] property

SDK docs URL.

Returns:

Type Description
Optional[str]

The flavor SDK docs URL.

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

Bases: BaseSandboxSettings

Local sandbox settings.

Source code in src/zenml/config/secret_reference_mixin.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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)

SandboxExecError

Bases: RuntimeError

Raised when a sandbox command fails to launch.

SandboxOutput(stdout: str, stderr: str, exit_code: int, stdout_truncated: bool = False, stderr_truncated: bool = False) dataclass

Result of fully consuming a sandbox process.

SandboxProcess(session: SandboxSession, started_at: float)

Bases: ABC

Handle to a command being executed inside a sandbox session.

Initialize the sandbox process.

Parameters:

Name Type Description Default
session SandboxSession

The owning session.

 required
started_at float

The wall-clock time the process started.

 required
Source code in src/zenml/sandboxes/process.py 
49
50
51
52
53
54
55
56
57
58
def __init__(self, session: "SandboxSession", started_at: float) -> None:
    """Initialize the sandbox process.

    Args:
        session: The owning session.
        started_at: The wall-clock time the process started.
    """
    self._session = session
    self._started_at = started_at
    self._collected = False
Attributes
exit_code: Optional[int] abstractmethod property

Exit code, or None if the command is still running.

Methods:
collect(*, max_chars: int = _DEFAULT_COLLECT_CHAR_COUNT) -> SandboxOutput

Wait for the process to exit and return the output.

Parameters:

Name Type Description Default
max_chars int

Maximum number of characters to collect per stream.

 _DEFAULT_COLLECT_CHAR_COUNT

Raises:

Type Description
RuntimeError

If the process output was already collected. The streams are drained by the first call, so a second call would fabricate an empty result instead of the real output.
drain_exception

Re-raised from a failed stdout or stderr drain.

Returns:

Type Description
SandboxOutput

The output of the process.
Source code in src/zenml/sandboxes/process.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def collect(
    self, *, max_chars: int = _DEFAULT_COLLECT_CHAR_COUNT
) -> SandboxOutput:
    """Wait for the process to exit and return the output.

    Args:
        max_chars: Maximum number of characters to collect per stream.

    Raises:
        RuntimeError: If the process output was already collected. The
            streams are drained by the first call, so a second call
            would fabricate an empty result instead of the real output.
        drain_exception: Re-raised from a failed stdout or stderr drain.

    Returns:
        The output of the process.
    """
    if self._collected:
        raise RuntimeError("process output was already collected")

    results: List[Optional[Tuple[str, bool]]] = [None, None]
    errors: List[Optional[BaseException]] = [None, None]

    def _drain(idx: int, stream: Iterator[str]) -> None:
        try:
            results[idx] = _drain_stream(stream, max_chars=max_chars)
        except BaseException as e:  # noqa: BLE001
            errors[idx] = e

    # Drain both streams concurrently to avoid deadlocks. Serial
    # draining deadlocks when a child writes more than one OS pipe
    # buffer (~64KB on Linux) to one stream before the other closes,
    # because the child blocks on write() while we're stuck on the
    # other read().
    stdout_drain_thread = threading.Thread(
        target=_drain, args=(0, self.stdout()), daemon=True
    )
    stderr_drain_thread = threading.Thread(
        target=_drain, args=(1, self.stderr()), daemon=True
    )
    stdout_drain_thread.start()
    stderr_drain_thread.start()
    stdout_drain_thread.join()
    stderr_drain_thread.join()

    # If a drain raised, the corresponding pipe is undrained. wait()
    # below would re-deadlock on a child still writing. Kill the
    # process first, then re-raise the original drain exception.
    drain_exception = errors[0] or errors[1]
    if drain_exception is not None:
        try:
            self.kill()
        except Exception:
            logger.debug(
                "kill() failed during drain-error cleanup",
                exc_info=True,
            )

        try:
            self.wait()
        except Exception:
            logger.debug(
                "wait() failed during drain-error cleanup",
                exc_info=True,
            )

        raise drain_exception

    stdout, stdout_truncated = results[0] or ("", False)
    stderr, stderr_truncated = results[1] or ("", False)

    exit_code = self.wait()
    self._collected = True
    self._session._log_exec_result(
        exit_code=exit_code, started_at=self._started_at
    )

    return SandboxOutput(
        stdout=stdout,
        stderr=stderr,
        exit_code=exit_code,
        stdout_truncated=stdout_truncated,
        stderr_truncated=stderr_truncated,
    )
kill() -> None abstractmethod

Terminates the process.

Source code in src/zenml/sandboxes/process.py 
87
88
89
@abstractmethod
def kill(self) -> None:
    """Terminates the process."""
stderr() -> Iterator[str] abstractmethod

Yield stderr lines.

Returns:

Type Description
Iterator[str]

An iterator over standard error lines.
Source code in src/zenml/sandboxes/process.py 
68
69
70
71
72
73
74
@abstractmethod
def stderr(self) -> Iterator[str]:
    """Yield stderr lines.

    Returns:
        An iterator over standard error lines.
    """
stdout() -> Iterator[str] abstractmethod

Yield stdout lines.

Returns:

Type Description
Iterator[str]

An iterator over standard output lines.
Source code in src/zenml/sandboxes/process.py 
60
61
62
63
64
65
66
@abstractmethod
def stdout(self) -> Iterator[str]:
    """Yield stdout lines.

    Returns:
        An iterator over standard output lines.
    """
wait(timeout: Optional[float] = None) -> int abstractmethod

Blocks until the process exits.

Parameters:

Name Type Description Default
timeout Optional[float]

Timeout in seconds to wait.

 None

Returns:

Type Description
int

The exit code.
Source code in src/zenml/sandboxes/process.py 
76
77
78
79
80
81
82
83
84
85
@abstractmethod
def wait(self, timeout: Optional[float] = None) -> int:
    """Blocks until the process exits.

    Args:
        timeout: Timeout in seconds to wait.

    Returns:
        The exit code.
    """

SandboxSession(*, id: str, parent: BaseSandbox, destroy_on_exit: bool = False)

Bases: ABC

Sandbox session.

Initialize the sandbox session.

Parameters:

Name Type Description Default
id str

Session identifier.

 required
parent BaseSandbox

The sandbox component that created this session.

 required
destroy_on_exit bool

Whether to destroy the sandbox session when the session context manager exits.

 False
Source code in src/zenml/sandboxes/session.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    *,
    id: str,
    parent: "BaseSandbox",
    destroy_on_exit: bool = False,
) -> None:
    """Initialize the sandbox session.

    Args:
        id: Session identifier.
        parent: The sandbox component that created this session.
        destroy_on_exit: Whether to destroy the sandbox session when the
            session context manager exits.
    """
    self.id = id
    self._parent = parent
    self._destroy_on_exit = destroy_on_exit
    self._closed = False
    self._logging_context: Optional["LoggingContext"] = None
    self._logging_disabled = False
    self._logging_lock = threading.Lock()
    self._publish_sandbox_metadata()
Attributes
closed: bool property

Whether this session handle has been closed.

Returns:

Type Description
bool

True once close() or destroy() has been called.
Methods:
aexec(command: Union[str, List[str]], *, cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None) -> SandboxProcess async

Async execute a command in the sandbox session.

Parameters:

Name Type Description Default
command Union[str, List[str]]

The command to execute.

 required
cwd Optional[str]

Optional working directory override.

 None
env Optional[Dict[str, str]]

Optional environment variables to set in the environment executing the command.

 None

Raises:

Type Description
NotImplementedError

If the sandbox does not support async exec.

Returns:

Type Description
SandboxProcess

Process handle.
Source code in src/zenml/sandboxes/session.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
async def aexec(
    self,
    command: Union[str, List[str]],
    *,
    cwd: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
) -> SandboxProcess:
    """Async execute a command in the sandbox session.

    Args:
        command: The command to execute.
        cwd: Optional working directory override.
        env: Optional environment variables to set in the environment
            executing the command.

    Raises:
        NotImplementedError: If the sandbox does not support async exec.

    Returns:
        Process handle.
    """
    self._ensure_open()
    raise NotImplementedError(
        f"{type(self).__name__} does not support async execution."
    )
close() -> None

Close the sandbox session handle. Terminal and idempotent.

After closing, the session handle rejects further use (exec, snapshots, file transfer) — re-attach to the sandbox for a fresh handle. For remote sandboxes, closing does not terminate the sandbox on the provider: it keeps running until its TTL expires or destroy() is called.

Source code in src/zenml/sandboxes/session.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def close(self) -> None:
    """Close the sandbox session handle. Terminal and idempotent.

    After closing, the session handle rejects further use (`exec`,
    snapshots, file transfer) — re-attach to the sandbox for a fresh
    handle. For remote sandboxes, closing does not terminate the
    sandbox on the provider: it keeps running until its TTL expires
    or `destroy()` is called.
    """
    if self._closed:
        return
    self._closed = True
    try:
        self._close()
    finally:
        self._close_logging_context()
create_snapshot() -> SandboxSnapshot

Create a snapshot of the sandbox session.

Returns:

Type Description
SandboxSnapshot

A sandbox snapshot.
Source code in src/zenml/sandboxes/session.py 
163
164
165
166
167
168
169
170
def create_snapshot(self) -> SandboxSnapshot:
    """Create a snapshot of the sandbox session.

    Returns:
        A sandbox snapshot.
    """
    self._ensure_open()
    return self._create_snapshot()
destroy() -> None

Destroy the sandbox session. Terminal like close().

Terminates the sandbox on the provider and then closes this handle, so after a successful call it is no longer possible to attach to the session. Flavor failures (including NotImplementedError from flavors without destroy support) propagate and leave the handle open for retry, because the flavor hook runs before close(). Callable on an already-closed handle: close() is idempotent.

Source code in src/zenml/sandboxes/session.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
def destroy(self) -> None:
    """Destroy the sandbox session. Terminal like `close()`.

    Terminates the sandbox on the provider and then closes this
    handle, so after a successful call it is no longer possible to
    attach to the session. Flavor failures (including
    `NotImplementedError` from flavors without destroy support)
    propagate and leave the handle open for retry, because the
    flavor hook runs before `close()`. Callable on an
    already-closed handle: `close()` is idempotent.
    """
    self._destroy()
    self.close()
download_file(remote_path: str, local_path: str) -> None

Download a file from the sandbox to the local filesystem.

Parameters:

Name Type Description Default
remote_path str

Source path in the sandbox.

 required
local_path str

Destination path on the caller's filesystem.

 required
Source code in src/zenml/sandboxes/session.py 
210
211
212
213
214
215
216
217
218
def download_file(self, remote_path: str, local_path: str) -> None:
    """Download a file from the sandbox to the local filesystem.

    Args:
        remote_path: Source path in the sandbox.
        local_path: Destination path on the caller's filesystem.
    """
    self._ensure_open()
    self._download_file(remote_path, local_path)
exec(command: Union[str, List[str]], *, cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None) -> SandboxProcess

Execute a command in the sandbox session.

Parameters:

Name Type Description Default
command Union[str, List[str]]

The command to execute.

 required
cwd Optional[str]

Optional working directory override.

 None
env Optional[Dict[str, str]]

Optional environment variables to set in the environment executing the command.

 None

Returns:

Type Description
SandboxProcess

Process handle.
Source code in src/zenml/sandboxes/session.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def exec(
    self,
    command: Union[str, List[str]],
    *,
    cwd: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
) -> SandboxProcess:
    """Execute a command in the sandbox session.

    Args:
        command: The command to execute.
        cwd: Optional working directory override.
        env: Optional environment variables to set in the environment
            executing the command.

    Returns:
        Process handle.
    """
    self._ensure_open()
    return self._exec(command, cwd=cwd, env=env)
upload_file(local_path: str, remote_path: str) -> None

Upload a file to the sandbox session.

Parameters:

Name Type Description Default
local_path str

Source path on the caller's filesystem.

 required
remote_path str

Destination path in the sandbox.

 required
Source code in src/zenml/sandboxes/session.py 
186
187
188
189
190
191
192
193
194
def upload_file(self, local_path: str, remote_path: str) -> None:
    """Upload a file to the sandbox session.

    Args:
        local_path: Source path on the caller's filesystem.
        remote_path: Destination path in the sandbox.
    """
    self._ensure_open()
    self._upload_file(local_path, remote_path)

SandboxSessionClosedError

Bases: RuntimeError

Raised when a closed sandbox session is used.

SandboxSnapshot

Bases: BaseModel

Sandbox snapshot.

Modules

base

Base sandbox flavor and component.

Classes
BaseSandbox(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 sandbox components.

Source code in src/zenml/stack/stack_component.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
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: BaseSandboxConfig property

The sandbox component configuration.

Returns:

Type Description
BaseSandboxConfig

The sandbox component configuration.
settings_class: Optional[Type[BaseSettings]] property

The sandbox settings class.

Returns:

Type Description
Optional[Type[BaseSettings]]

The sandbox settings class.
Methods:
attach(session_id: str) -> SandboxSession

Attach to a running sandbox session.

Parameters:

Name Type Description Default
session_id str

The ID of the running sandbox session.

 required

Raises:

Type Description
NotImplementedError

If the sandbox does not support attaching to a running sandbox session.

Returns:

Type Description
SandboxSession

Sandbox session.
Source code in src/zenml/sandboxes/base.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def attach(self, session_id: str) -> SandboxSession:
    """Attach to a running sandbox session.

    Args:
        session_id: The ID of the running sandbox session.

    Raises:
        NotImplementedError: If the sandbox does not support attaching to a
            running sandbox session.

    Returns:
        Sandbox session.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support attaching to a running "
        "sandbox session."
    )
create_session(settings: Optional[BaseSandboxSettings] = None, destroy_on_exit: bool = False) -> SandboxSession abstractmethod

Create a fresh sandbox session.

Parameters:

Name Type Description Default
settings Optional[BaseSandboxSettings]

The sandbox settings.

 None
destroy_on_exit bool

Whether to destroy the sandbox session when the session context manager exits.

 False

Returns:

Type Description
SandboxSession

A new sandbox session.
Source code in src/zenml/sandboxes/base.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@abstractmethod
def create_session(
    self,
    settings: Optional[BaseSandboxSettings] = None,
    destroy_on_exit: bool = False,
) -> SandboxSession:
    """Create a fresh sandbox session.

    Args:
        settings: The sandbox settings.
        destroy_on_exit: Whether to destroy the sandbox session when the
            session context manager exits.

    Returns:
        A new sandbox session.
    """
resolve_settings(override: Optional[BaseSandboxSettings] = None) -> BaseSandboxSettings

Resolve the sandbox settings.

Parameters:

Name Type Description Default
override Optional[BaseSandboxSettings]

Per-call settings override. If provided these will override values in the component settings/config.

 None

Returns:

Type Description
BaseSandboxSettings

The resolved settings.
Source code in src/zenml/sandboxes/base.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def resolve_settings(
    self, override: Optional[BaseSandboxSettings] = None
) -> BaseSandboxSettings:
    """Resolve the sandbox settings.

    Args:
        override: Per-call settings override. If provided these will
            override values in the component settings/config.

    Returns:
        The resolved settings.
    """
    from zenml.steps.step_context import StepContext

    if step_context := StepContext.get():
        settings_dict = self.get_settings(
            step_context.step_run
        ).model_dump(exclude_unset=True)
    else:
        settings_dict = self.config.model_dump(exclude_unset=True)

    if override is not None:
        settings_dict.update(override.model_dump(exclude_unset=True))

    assert self.settings_class is not None
    return cast(
        BaseSandboxSettings,
        self.settings_class.model_validate(settings_dict),
    )
restore(snapshot: SandboxSnapshot) -> SandboxSession

Restore a sandbox session from a snapshot.

Parameters:

Name Type Description Default
snapshot SandboxSnapshot

The snapshot to restore from.

 required

Raises:

Type Description
NotImplementedError

If the sandbox does not support restoring from a snapshot.

Returns:

Type Description
SandboxSession

Sandbox session.
Source code in src/zenml/sandboxes/base.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def restore(self, snapshot: SandboxSnapshot) -> SandboxSession:
    """Restore a sandbox session from a snapshot.

    Args:
        snapshot: The snapshot to restore from.

    Raises:
        NotImplementedError: If the sandbox does not support restoring from
            a snapshot.

    Returns:
        Sandbox session.
    """
    raise NotImplementedError(
        f"{type(self).__name__} does not support restoring from a snapshot."
    )
BaseSandboxConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: StackComponentConfig

Sandbox configuration.

Source code in src/zenml/stack/stack_component.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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

Whether this sandbox component is running locally.

Returns:

Type Description
bool

Whether this sandbox component is running locally.
BaseSandboxFlavor

Bases: Flavor

Base sandbox flavor.

Attributes
config_class: Type[BaseSandboxConfig] property

Configuration class.

Returns:

Type Description
Type[BaseSandboxConfig]

The configuration class.
implementation_class: Type[BaseSandbox] abstractmethod property

Implementation class.

Returns:

Type Description
Type[BaseSandbox]

The implementation class.
type: StackComponentType property

Stack component type.

Returns:

Type Description
StackComponentType

StackComponentType.SANDBOX.
BaseSandboxSettings(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseSettings

Sandbox settings.

Source code in src/zenml/config/secret_reference_mixin.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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)
ContainerizedSandboxSettings(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseSandboxSettings

Containerized sandbox settings.

Source code in src/zenml/config/secret_reference_mixin.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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:

docker_sandbox

Docker sandbox flavor.

Classes
DockerSandbox(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: BaseSandbox

Docker daemon-backed sandbox.

Source code in src/zenml/stack/stack_component.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
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: DockerSandboxConfig property

Docker sandbox configuration.

Returns:

Type Description
DockerSandboxConfig

The Docker sandbox configuration.
docker_client: DockerClient property

Initialize and/or return the docker client.

Returns:

Type Description
DockerClient

The docker client.
docker_engine: DockerContainerEngine property

Initialize and/or return the docker engine.

Returns:

Type Description
DockerContainerEngine

The docker engine.
settings_class: Optional[Type[BaseSandboxSettings]] property

Settings class.

Returns:

Type Description
Optional[Type[BaseSandboxSettings]]

DockerSandboxSettings.
Methods:
attach(session_id: str) -> SandboxSession

Attach to a running Docker sandbox session.

Parameters:

Name Type Description Default
session_id str

The ID of the running sandbox session.

 required

Raises:

Type Description
KeyError

If no running session with the given ID exists.

Returns:

Type Description
SandboxSession

Sandbox session.
Source code in src/zenml/sandboxes/docker_sandbox.py 
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
def attach(self, session_id: str) -> SandboxSession:
    """Attach to a running Docker sandbox session.

    Args:
        session_id: The ID of the running sandbox session.

    Raises:
        KeyError: If no running session with the given ID exists.

    Returns:
        Sandbox session.
    """
    containers = self.docker_client.containers.list(
        filters={"label": f"{_SESSION_ID_LABEL}={session_id}"}
    )
    if not containers:
        raise KeyError(
            f"No running Docker sandbox session `{session_id}` found."
        )

    settings = cast(DockerSandboxSettings, self.resolve_settings())
    return DockerSandboxSession(
        container=containers[0],
        workdir=settings.workdir,
        env=self._resolve_session_environment(settings),
        parent=self,
        id=session_id,
    )
create_session(settings: Optional[BaseSandboxSettings] = None, destroy_on_exit: bool = False) -> SandboxSession

Create a Docker sandbox session backed by a new container.

Parameters:

Name Type Description Default
settings Optional[BaseSandboxSettings]

Optional settings overrides.

 None
destroy_on_exit bool

Whether to destroy the sandbox session when the session context manager exits.

 False

Returns:

Type Description
SandboxSession

A Docker sandbox session.
Source code in src/zenml/sandboxes/docker_sandbox.py 
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
def create_session(
    self,
    settings: Optional[BaseSandboxSettings] = None,
    destroy_on_exit: bool = False,
) -> SandboxSession:
    """Create a Docker sandbox session backed by a new container.

    Args:
        settings: Optional settings overrides.
        destroy_on_exit: Whether to destroy the sandbox session when the
            session context manager exits.

    Returns:
        A Docker sandbox session.
    """
    settings = cast(DockerSandboxSettings, self.resolve_settings(settings))
    self._ensure_image(settings)

    session_id = f"docker-{uuid4().hex[:12]}"

    run_args = copy.deepcopy(settings.run_args)
    labels = run_args.pop("labels", {})
    labels[_SESSION_ID_LABEL] = session_id
    if settings.cpu_limit is not None:
        run_args["nano_cpus"] = int(settings.cpu_limit * 1e9)
    if settings.memory_limit is not None:
        run_args["mem_limit"] = settings.memory_limit

    container = self.docker_client.containers.run(
        image=settings.image,
        command=["sleep", "infinity"],
        entrypoint=[],
        detach=True,
        name=f"zenml-sandbox-{session_id}",
        labels=labels,
        working_dir=settings.workdir,
        **run_args,
    )

    return DockerSandboxSession(
        container=container,
        workdir=settings.workdir,
        env=self._resolve_session_environment(settings),
        parent=self,
        destroy_on_exit=destroy_on_exit,
        id=session_id,
    )
restore(snapshot: SandboxSnapshot) -> SandboxSession

Restore a sandbox session from a snapshot.

Parameters:

Name Type Description Default
snapshot SandboxSnapshot

The snapshot to restore from.

 required

Returns:

Type Description
SandboxSession

A new session started from the snapshot image.
Source code in src/zenml/sandboxes/docker_sandbox.py 
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
def restore(self, snapshot: SandboxSnapshot) -> SandboxSession:
    """Restore a sandbox session from a snapshot.

    Args:
        snapshot: The snapshot to restore from.

    Returns:
        A new session started from the snapshot image.
    """
    self._validate_snapshot(snapshot)
    # Snapshot images are stored locally, so they must never be pulled.
    return self.create_session(
        settings=DockerSandboxSettings(
            image=snapshot.ref,
            pull_policy=DockerSandboxPullPolicy.NEVER,
        )
    )
DockerSandboxConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseSandboxConfig, DockerSandboxSettings

Docker sandbox configuration.

Source code in src/zenml/stack/stack_component.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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

Whether this sandbox component is running locally.

Returns:

Type Description
bool

Whether this sandbox component is running locally.
DockerSandboxFlavor

Bases: BaseSandboxFlavor

Docker sandbox flavor.

Attributes
config_class: Type[DockerSandboxConfig] property

Config class.

Returns:

Type Description
Type[DockerSandboxConfig]

DockerSandboxConfig.
docs_url: Optional[str] property

Docs URL.

Returns:

Type Description
Optional[str]

The docs URL.
implementation_class: Type[DockerSandbox] property

Implementation class.

Returns:

Type Description
Type[DockerSandbox]

DockerSandbox.
logo_url: str property

Dashboard logo URL.

Returns:

Type Description
str

The flavor logo URL.
name: str property

Flavor name.

Returns:

Type Description
str

The flavor name.
sdk_docs_url: Optional[str] property

SDK docs URL.

Returns:

Type Description
Optional[str]

The flavor SDK docs URL.
DockerSandboxProcess(client: APIClient, container_id: str, exec_id: str, pid_file: str, session: DockerSandboxSession, started_at: float)

Bases: SandboxProcess

Docker sandbox process wrapping an exec instance.

Initialize the Docker sandbox process.

Parameters:

Name Type Description Default
client APIClient

Low-level Docker API client.

 required
container_id str

ID of the container the command runs in.

 required
exec_id str

ID of the exec instance running the command.

 required
pid_file str

Path of the file inside the container that the command pid was recorded to.

 required
session DockerSandboxSession

The owning session.

 required
started_at float

The wall-clock time the process started.

 required
Source code in src/zenml/sandboxes/docker_sandbox.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def __init__(
    self,
    client: APIClient,
    container_id: str,
    exec_id: str,
    pid_file: str,
    session: "DockerSandboxSession",
    started_at: float,
) -> None:
    """Initialize the Docker sandbox process.

    Args:
        client: Low-level Docker API client.
        container_id: ID of the container the command runs in.
        exec_id: ID of the exec instance running the command.
        pid_file: Path of the file inside the container that the
            command pid was recorded to.
        session: The owning session.
        started_at: The wall-clock time the process started.
    """
    super().__init__(session=session, started_at=started_at)
    self._client = client
    self._container_id = container_id
    self._exec_id = exec_id
    self._pid_file = pid_file
    self._exit_code: Optional[int] = None
    self._stdout_queue: "queue.Queue[object]" = queue.Queue()
    self._stderr_queue: "queue.Queue[object]" = queue.Queue()
    self._done = threading.Event()
    self._drain_thread = threading.Thread(target=self._drain, daemon=True)
    self._drain_thread.start()
Attributes
exit_code: Optional[int] property

Exit code, or None if the command is still running.

Returns:

Type Description
Optional[int]

The exit code or None.
Methods:
kill() -> None

Terminate the command by killing its recorded pid.

Source code in src/zenml/sandboxes/docker_sandbox.py 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def kill(self) -> None:
    """Terminate the command by killing its recorded pid."""
    if self._done.is_set():
        return

    try:
        kill_exec = self._client.exec_create(
            self._container_id,
            ["sh", "-c", f'kill -9 "$(cat {self._pid_file})"'],
        )
        self._client.exec_start(kill_exec["Id"])
    except Exception as e:
        logger.warning(
            "DockerSandbox kill() failed: %s",
            e,
            exc_info=True,
        )
stderr() -> Iterator[str]

Stderr line iterator.

Returns:

Type Description
Iterator[str]

Stderr line iterator.
Source code in src/zenml/sandboxes/docker_sandbox.py 
272
273
274
275
276
277
278
279
280
def stderr(self) -> Iterator[str]:
    """Stderr line iterator.

    Returns:
        Stderr line iterator.
    """
    return self._session._wrap_stream(
        self._iter_queue(self._stderr_queue), log_level=logging.ERROR
    )
stdout() -> Iterator[str]

Stdout line iterator.

Returns:

Type Description
Iterator[str]

Stdout line iterator.
Source code in src/zenml/sandboxes/docker_sandbox.py 
262
263
264
265
266
267
268
269
270
def stdout(self) -> Iterator[str]:
    """Stdout line iterator.

    Returns:
        Stdout line iterator.
    """
    return self._session._wrap_stream(
        self._iter_queue(self._stdout_queue), log_level=logging.INFO
    )
wait(timeout: Optional[float] = None) -> int

Wait for command completion.

Parameters:

Name Type Description Default
timeout Optional[float]

Timeout in seconds to wait.

 None

Raises:

Type Description
TimeoutError

If the command did not finish in time.

Returns:

Type Description
int

The exit code.
Source code in src/zenml/sandboxes/docker_sandbox.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def wait(self, timeout: Optional[float] = None) -> int:
    """Wait for command completion.

    Args:
        timeout: Timeout in seconds to wait.

    Raises:
        TimeoutError: If the command did not finish in time.

    Returns:
        The exit code.
    """
    if not self._done.wait(timeout):
        raise TimeoutError(
            "Timed out waiting for Docker sandbox command to finish."
        )

    assert self._exit_code is not None
    return self._exit_code
DockerSandboxPullPolicy

Bases: StrEnum

Docker sandbox image pull policy.

DockerSandboxSession(container: Container, workdir: str, env: Dict[str, str], *, parent: BaseSandbox, destroy_on_exit: bool = False, id: str)

Bases: SandboxSession

Docker sandbox session backed by a container.

Initialize the Docker sandbox session.

Parameters:

Name Type Description Default
container Container

The running container backing this session.

 required
workdir str

Working directory inside the container.

 required
env Dict[str, str]

Environment variables to set for this session.

 required
parent BaseSandbox

The sandbox component that created this session.

 required
destroy_on_exit bool

Whether to destroy the sandbox session when the session context manager exits.

 False
id str

Session ID.

 required
Source code in src/zenml/sandboxes/docker_sandbox.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
def __init__(
    self,
    container: "Container",
    workdir: str,
    env: Dict[str, str],
    *,
    parent: "BaseSandbox",
    destroy_on_exit: bool = False,
    id: str,
) -> None:
    """Initialize the Docker sandbox session.

    Args:
        container: The running container backing this session.
        workdir: Working directory inside the container.
        env: Environment variables to set for this session.
        parent: The sandbox component that created this session.
        destroy_on_exit: Whether to destroy the sandbox session when the
            session context manager exits.
        id: Session ID.
    """
    self._container = container
    self._workdir = workdir
    self._env = env
    super().__init__(
        id=id,
        parent=parent,
        destroy_on_exit=destroy_on_exit,
    )
Methods:
DockerSandboxSettings(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: ContainerizedSandboxSettings

Docker sandbox settings.

Source code in src/zenml/config/secret_reference_mixin.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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:

local_sandbox

Local sandbox flavor.

Classes
LocalSandbox(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: BaseSandbox

Local subprocess-based Sandbox.

Warning: This class does NOT provide any isolation and is not suitable for running untrusted code.

Source code in src/zenml/stack/stack_component.py 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
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: LocalSandboxConfig property

Local sandbox configuration.

Returns:

Type Description
LocalSandboxConfig

The local sandbox configuration.
settings_class: Optional[Type[BaseSandboxSettings]] property

Settings class.

Returns:

Type Description
Optional[Type[BaseSandboxSettings]]

LocalSandboxSettings.
Methods:
create_session(settings: Optional[BaseSandboxSettings] = None, destroy_on_exit: bool = False) -> SandboxSession

Create a local sandbox session in a clean working directory.

Parameters:

Name Type Description Default
settings Optional[BaseSandboxSettings]

Optional settings overrides.

 None
destroy_on_exit bool

Whether to destroy the sandbox session when the session context manager exits.

 False

Returns:

Type Description
SandboxSession

A local sandbox session.
Source code in src/zenml/sandboxes/local_sandbox.py 
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
def create_session(
    self,
    settings: Optional[BaseSandboxSettings] = None,
    destroy_on_exit: bool = False,
) -> SandboxSession:
    """Create a local sandbox session in a clean working directory.

    Args:
        settings: Optional settings overrides.
        destroy_on_exit: Whether to destroy the sandbox session when the
            session context manager exits.

    Returns:
        A local sandbox session.
    """
    logger.warning(
        "The local sandbox does not provide any isolation. Any code "
        "executed in this sandbox will have full access to the local "
        "filesystem and network."
    )

    settings = cast(LocalSandboxSettings, self.resolve_settings(settings))
    env = self._resolve_session_environment(settings)
    workdir = tempfile.mkdtemp(prefix="zenml-local-sandbox-")

    return LocalSandboxSession(
        workdir=workdir,
        env=env,
        parent=self,
        destroy_on_exit=destroy_on_exit,
    )
LocalSandboxConfig(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseSandboxConfig, LocalSandboxSettings

Local sandbox configuration.

Source code in src/zenml/stack/stack_component.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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)
LocalSandboxFlavor

Bases: BaseSandboxFlavor

Local subprocess sandbox flavor.

Attributes
config_class: Type[LocalSandboxConfig] property

Config class.

Returns:

Type Description
Type[LocalSandboxConfig]

LocalSandboxConfig.
docs_url: Optional[str] property

Docs URL.

Returns:

Type Description
Optional[str]

The docs URL.
implementation_class: Type[LocalSandbox] property

Implementation class.

Returns:

Type Description
Type[LocalSandbox]

LocalSandbox.
logo_url: str property

Dashboard logo URL.

Returns:

Type Description
str

The flavor logo URL.
name: str property

Flavor name.

Returns:

Type Description
str

The flavor name.
sdk_docs_url: Optional[str] property

SDK docs URL.

Returns:

Type Description
Optional[str]

The flavor SDK docs URL.
LocalSandboxProcess(process: subprocess.Popen[str], session: LocalSandboxSession, started_at: float)

Bases: SandboxProcess

Local sandbox process wrapping a subprocess.

Initialize the local sandbox process.

Parameters:

Name Type Description Default
process Popen[str]

The subprocess handle.

 required
session LocalSandboxSession

The owning session.

 required
started_at float

The wall-clock time the launch began.

 required
Source code in src/zenml/sandboxes/local_sandbox.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(
    self,
    process: "subprocess.Popen[str]",
    session: "LocalSandboxSession",
    started_at: float,
) -> None:
    """Initialize the local sandbox process.

    Args:
        process: The subprocess handle.
        session: The owning session.
        started_at: The wall-clock time the launch began.
    """
    super().__init__(session=session, started_at=started_at)
    self._process = process
Attributes
exit_code: Optional[int] property

Exit code, or None if the subprocess is still running.

Returns:

Type Description
Optional[int]

The exit code or None.
Methods:
kill() -> None

Kill the subprocess by sending SIGKILL.

Source code in src/zenml/sandboxes/local_sandbox.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
def kill(self) -> None:
    """Kill the subprocess by sending SIGKILL."""
    try:
        self._process.kill()
    except ProcessLookupError:
        pass
    except Exception as e:
        logger.warning(
            "LocalSandbox kill() failed: %s. Process may still be "
            "running.",
            e,
            exc_info=True,
        )
stderr() -> Iterator[str]

Stderr line iterator.

Returns:

Type Description
Iterator[str]

Stderr line iterator.
Source code in src/zenml/sandboxes/local_sandbox.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def stderr(self) -> Iterator[str]:
    """Stderr line iterator.

    Returns:
        Stderr line iterator.
    """

    def _stderr() -> Iterator[str]:
        if self._process.stderr is None:
            return

        for line in self._process.stderr:
            yield line

    return self._session._wrap_stream(_stderr(), log_level=logging.ERROR)
stdout() -> Iterator[str]

Stdout line iterator.

Returns:

Type Description
Iterator[str]

Stdout line iterator.
Source code in src/zenml/sandboxes/local_sandbox.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def stdout(self) -> Iterator[str]:
    """Stdout line iterator.

    Returns:
        Stdout line iterator.
    """

    def _stdout() -> Iterator[str]:
        if self._process.stdout is None:
            return

        for line in self._process.stdout:
            yield line

    return self._session._wrap_stream(_stdout(), log_level=logging.INFO)
wait(timeout: Optional[float] = None) -> int

Block until the subprocess exits.

Parameters:

Name Type Description Default
timeout Optional[float]

Timeout in seconds to wait.

 None

Returns:

Type Description
int

The exit code.
Source code in src/zenml/sandboxes/local_sandbox.py 
141
142
143
144
145
146
147
148
149
150
def wait(self, timeout: Optional[float] = None) -> int:
    """Block until the subprocess exits.

    Args:
        timeout: Timeout in seconds to wait.

    Returns:
        The exit code.
    """
    return self._process.wait(timeout=timeout)
LocalSandboxSession(workdir: str, env: Dict[str, str], *, parent: BaseSandbox, destroy_on_exit: bool = False)

Bases: SandboxSession

Local sandbox session.

Initialize the local sandbox session.

Parameters:

Name Type Description Default
workdir str

Absolute path to the working directory for this session. This directory is cleaned up when the session is closed.

 required
env Dict[str, str]

Environment variables to set for this session.

 required
parent BaseSandbox

The sandbox component that created this session.

 required
destroy_on_exit bool

Whether to destroy the sandbox session when the session context manager exits.

 False
Source code in src/zenml/sandboxes/local_sandbox.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def __init__(
    self,
    workdir: str,
    env: Dict[str, str],
    *,
    parent: "BaseSandbox",
    destroy_on_exit: bool = False,
) -> None:
    """Initialize the local sandbox session.

    Args:
        workdir: Absolute path to the working directory for this session.
            This directory is cleaned up when the session is closed.
        env: Environment variables to set for this session.
        parent: The sandbox component that created this session.
        destroy_on_exit: Whether to destroy the sandbox session when the
            session context manager exits.
    """
    super().__init__(
        id=f"local-{uuid.uuid4().hex[:12]}",
        parent=parent,
        destroy_on_exit=destroy_on_exit,
    )
    self._workdir = workdir
    self._env = env
    self._processes: List["subprocess.Popen[str]"] = []
Methods:
LocalSandboxSettings(warn_about_plain_text_secrets: bool = False, **kwargs: Any)

Bases: BaseSandboxSettings

Local sandbox settings.

Source code in src/zenml/config/secret_reference_mixin.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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:

process

Sandbox process.

Classes
SandboxExecError

Bases: RuntimeError

Raised when a sandbox command fails to launch.

SandboxOutput(stdout: str, stderr: str, exit_code: int, stdout_truncated: bool = False, stderr_truncated: bool = False) dataclass

Result of fully consuming a sandbox process.

SandboxProcess(session: SandboxSession, started_at: float)

Bases: ABC

Handle to a command being executed inside a sandbox session.

Initialize the sandbox process.

Parameters:

Name Type Description Default
session SandboxSession

The owning session.

 required
started_at float

The wall-clock time the process started.

 required
Source code in src/zenml/sandboxes/process.py 
49
50
51
52
53
54
55
56
57
58
def __init__(self, session: "SandboxSession", started_at: float) -> None:
    """Initialize the sandbox process.

    Args:
        session: The owning session.
        started_at: The wall-clock time the process started.
    """
    self._session = session
    self._started_at = started_at
    self._collected = False
Attributes
exit_code: Optional[int] abstractmethod property

Exit code, or None if the command is still running.

Methods:
collect(*, max_chars: int = _DEFAULT_COLLECT_CHAR_COUNT) -> SandboxOutput

Wait for the process to exit and return the output.

Parameters:

Name Type Description Default
max_chars int

Maximum number of characters to collect per stream.

 _DEFAULT_COLLECT_CHAR_COUNT

Raises:

Type Description
RuntimeError

If the process output was already collected. The streams are drained by the first call, so a second call would fabricate an empty result instead of the real output.
drain_exception

Re-raised from a failed stdout or stderr drain.

Returns:

Type Description
SandboxOutput

The output of the process.
Source code in src/zenml/sandboxes/process.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def collect(
    self, *, max_chars: int = _DEFAULT_COLLECT_CHAR_COUNT
) -> SandboxOutput:
    """Wait for the process to exit and return the output.

    Args:
        max_chars: Maximum number of characters to collect per stream.

    Raises:
        RuntimeError: If the process output was already collected. The
            streams are drained by the first call, so a second call
            would fabricate an empty result instead of the real output.
        drain_exception: Re-raised from a failed stdout or stderr drain.

    Returns:
        The output of the process.
    """
    if self._collected:
        raise RuntimeError("process output was already collected")

    results: List[Optional[Tuple[str, bool]]] = [None, None]
    errors: List[Optional[BaseException]] = [None, None]

    def _drain(idx: int, stream: Iterator[str]) -> None:
        try:
            results[idx] = _drain_stream(stream, max_chars=max_chars)
        except BaseException as e:  # noqa: BLE001
            errors[idx] = e

    # Drain both streams concurrently to avoid deadlocks. Serial
    # draining deadlocks when a child writes more than one OS pipe
    # buffer (~64KB on Linux) to one stream before the other closes,
    # because the child blocks on write() while we're stuck on the
    # other read().
    stdout_drain_thread = threading.Thread(
        target=_drain, args=(0, self.stdout()), daemon=True
    )
    stderr_drain_thread = threading.Thread(
        target=_drain, args=(1, self.stderr()), daemon=True
    )
    stdout_drain_thread.start()
    stderr_drain_thread.start()
    stdout_drain_thread.join()
    stderr_drain_thread.join()

    # If a drain raised, the corresponding pipe is undrained. wait()
    # below would re-deadlock on a child still writing. Kill the
    # process first, then re-raise the original drain exception.
    drain_exception = errors[0] or errors[1]
    if drain_exception is not None:
        try:
            self.kill()
        except Exception:
            logger.debug(
                "kill() failed during drain-error cleanup",
                exc_info=True,
            )

        try:
            self.wait()
        except Exception:
            logger.debug(
                "wait() failed during drain-error cleanup",
                exc_info=True,
            )

        raise drain_exception

    stdout, stdout_truncated = results[0] or ("", False)
    stderr, stderr_truncated = results[1] or ("", False)

    exit_code = self.wait()
    self._collected = True
    self._session._log_exec_result(
        exit_code=exit_code, started_at=self._started_at
    )

    return SandboxOutput(
        stdout=stdout,
        stderr=stderr,
        exit_code=exit_code,
        stdout_truncated=stdout_truncated,
        stderr_truncated=stderr_truncated,
    )
kill() -> None abstractmethod

Terminates the process.

Source code in src/zenml/sandboxes/process.py 
87
88
89
@abstractmethod
def kill(self) -> None:
    """Terminates the process."""
stderr() -> Iterator[str] abstractmethod

Yield stderr lines.

Returns:

Type Description
Iterator[str]

An iterator over standard error lines.
Source code in src/zenml/sandboxes/process.py 
68
69
70
71
72
73
74
@abstractmethod
def stderr(self) -> Iterator[str]:
    """Yield stderr lines.

    Returns:
        An iterator over standard error lines.
    """
stdout() -> Iterator[str] abstractmethod

Yield stdout lines.

Returns:

Type Description
Iterator[str]

An iterator over standard output lines.
Source code in src/zenml/sandboxes/process.py 
60
61
62
63
64
65
66
@abstractmethod
def stdout(self) -> Iterator[str]:
    """Yield stdout lines.

    Returns:
        An iterator over standard output lines.
    """
wait(timeout: Optional[float] = None) -> int abstractmethod

Blocks until the process exits.

Parameters:

Name Type Description Default
timeout Optional[float]

Timeout in seconds to wait.

 None

Returns:

Type Description
int

The exit code.
Source code in src/zenml/sandboxes/process.py 
76
77
78
79
80
81
82
83
84
85
@abstractmethod
def wait(self, timeout: Optional[float] = None) -> int:
    """Blocks until the process exits.

    Args:
        timeout: Timeout in seconds to wait.

    Returns:
        The exit code.
    """
Functions:

session

Sandbox session.

Classes
SandboxSession(*, id: str, parent: BaseSandbox, destroy_on_exit: bool = False)

Bases: ABC

Sandbox session.

Initialize the sandbox session.

Parameters:

Name Type Description Default
id str

Session identifier.

 required
parent BaseSandbox

The sandbox component that created this session.

 required
destroy_on_exit bool

Whether to destroy the sandbox session when the session context manager exits.

 False
Source code in src/zenml/sandboxes/session.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    *,
    id: str,
    parent: "BaseSandbox",
    destroy_on_exit: bool = False,
) -> None:
    """Initialize the sandbox session.

    Args:
        id: Session identifier.
        parent: The sandbox component that created this session.
        destroy_on_exit: Whether to destroy the sandbox session when the
            session context manager exits.
    """
    self.id = id
    self._parent = parent
    self._destroy_on_exit = destroy_on_exit
    self._closed = False
    self._logging_context: Optional["LoggingContext"] = None
    self._logging_disabled = False
    self._logging_lock = threading.Lock()
    self._publish_sandbox_metadata()
Attributes
closed: bool property

Whether this session handle has been closed.

Returns:

Type Description
bool

True once close() or destroy() has been called.
Methods:
aexec(command: Union[str, List[str]], *, cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None) -> SandboxProcess async

Async execute a command in the sandbox session.

Parameters:

Name Type Description Default
command Union[str, List[str]]

The command to execute.

 required
cwd Optional[str]

Optional working directory override.

 None
env Optional[Dict[str, str]]

Optional environment variables to set in the environment executing the command.

 None

Raises:

Type Description
NotImplementedError

If the sandbox does not support async exec.

Returns:

Type Description
SandboxProcess

Process handle.
Source code in src/zenml/sandboxes/session.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
async def aexec(
    self,
    command: Union[str, List[str]],
    *,
    cwd: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
) -> SandboxProcess:
    """Async execute a command in the sandbox session.

    Args:
        command: The command to execute.
        cwd: Optional working directory override.
        env: Optional environment variables to set in the environment
            executing the command.

    Raises:
        NotImplementedError: If the sandbox does not support async exec.

    Returns:
        Process handle.
    """
    self._ensure_open()
    raise NotImplementedError(
        f"{type(self).__name__} does not support async execution."
    )
close() -> None

Close the sandbox session handle. Terminal and idempotent.

After closing, the session handle rejects further use (exec, snapshots, file transfer) — re-attach to the sandbox for a fresh handle. For remote sandboxes, closing does not terminate the sandbox on the provider: it keeps running until its TTL expires or destroy() is called.

Source code in src/zenml/sandboxes/session.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def close(self) -> None:
    """Close the sandbox session handle. Terminal and idempotent.

    After closing, the session handle rejects further use (`exec`,
    snapshots, file transfer) — re-attach to the sandbox for a fresh
    handle. For remote sandboxes, closing does not terminate the
    sandbox on the provider: it keeps running until its TTL expires
    or `destroy()` is called.
    """
    if self._closed:
        return
    self._closed = True
    try:
        self._close()
    finally:
        self._close_logging_context()
create_snapshot() -> SandboxSnapshot

Create a snapshot of the sandbox session.

Returns:

Type Description
SandboxSnapshot

A sandbox snapshot.
Source code in src/zenml/sandboxes/session.py 
163
164
165
166
167
168
169
170
def create_snapshot(self) -> SandboxSnapshot:
    """Create a snapshot of the sandbox session.

    Returns:
        A sandbox snapshot.
    """
    self._ensure_open()
    return self._create_snapshot()
destroy() -> None

Destroy the sandbox session. Terminal like close().

Terminates the sandbox on the provider and then closes this handle, so after a successful call it is no longer possible to attach to the session. Flavor failures (including NotImplementedError from flavors without destroy support) propagate and leave the handle open for retry, because the flavor hook runs before close(). Callable on an already-closed handle: close() is idempotent.

Source code in src/zenml/sandboxes/session.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
def destroy(self) -> None:
    """Destroy the sandbox session. Terminal like `close()`.

    Terminates the sandbox on the provider and then closes this
    handle, so after a successful call it is no longer possible to
    attach to the session. Flavor failures (including
    `NotImplementedError` from flavors without destroy support)
    propagate and leave the handle open for retry, because the
    flavor hook runs before `close()`. Callable on an
    already-closed handle: `close()` is idempotent.
    """
    self._destroy()
    self.close()
download_file(remote_path: str, local_path: str) -> None

Download a file from the sandbox to the local filesystem.

Parameters:

Name Type Description Default
remote_path str

Source path in the sandbox.

 required
local_path str

Destination path on the caller's filesystem.

 required
Source code in src/zenml/sandboxes/session.py 
210
211
212
213
214
215
216
217
218
def download_file(self, remote_path: str, local_path: str) -> None:
    """Download a file from the sandbox to the local filesystem.

    Args:
        remote_path: Source path in the sandbox.
        local_path: Destination path on the caller's filesystem.
    """
    self._ensure_open()
    self._download_file(remote_path, local_path)
exec(command: Union[str, List[str]], *, cwd: Optional[str] = None, env: Optional[Dict[str, str]] = None) -> SandboxProcess

Execute a command in the sandbox session.

Parameters:

Name Type Description Default
command Union[str, List[str]]

The command to execute.

 required
cwd Optional[str]

Optional working directory override.

 None
env Optional[Dict[str, str]]

Optional environment variables to set in the environment executing the command.

 None

Returns:

Type Description
SandboxProcess

Process handle.
Source code in src/zenml/sandboxes/session.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def exec(
    self,
    command: Union[str, List[str]],
    *,
    cwd: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
) -> SandboxProcess:
    """Execute a command in the sandbox session.

    Args:
        command: The command to execute.
        cwd: Optional working directory override.
        env: Optional environment variables to set in the environment
            executing the command.

    Returns:
        Process handle.
    """
    self._ensure_open()
    return self._exec(command, cwd=cwd, env=env)
upload_file(local_path: str, remote_path: str) -> None

Upload a file to the sandbox session.

Parameters:

Name Type Description Default
local_path str

Source path on the caller's filesystem.

 required
remote_path str

Destination path in the sandbox.

 required
Source code in src/zenml/sandboxes/session.py 
186
187
188
189
190
191
192
193
194
def upload_file(self, local_path: str, remote_path: str) -> None:
    """Upload a file to the sandbox session.

    Args:
        local_path: Source path on the caller's filesystem.
        remote_path: Destination path in the sandbox.
    """
    self._ensure_open()
    self._upload_file(local_path, remote_path)
SandboxSessionClosedError

Bases: RuntimeError

Raised when a closed sandbox session is used.

Functions:

snapshot

Sandbox snapshot.

Classes
SandboxSnapshot

Bases: BaseModel

Sandbox snapshot.