Skip to content

Sandboxes

zenml.sandboxes

Sandbox stack components for isolated code execution.

Attributes

LOCAL_SANDBOX_FLAVOR = 'local' module-attribute

__all__ = ['LOCAL_SANDBOX_FLAVOR', 'BaseSandbox', 'BaseSandboxConfig', 'BaseSandboxFlavor', 'BaseSandboxSettings', '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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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) -> SandboxSession abstractmethod

Create a fresh sandbox session.

Parameters:

Name Type Description Default
settings Optional[BaseSandboxSettings]

The sandbox settings.

 None

Returns:

Type Description
SandboxSession

A new sandbox session.
Source code in src/zenml/sandboxes/base.py 
75
76
77
78
79
80
81
82
83
84
85
86
@abstractmethod
def create_session(
    self, settings: Optional[BaseSandboxSettings] = None
) -> SandboxSession:
    """Create a fresh sandbox session.

    Args:
        settings: The sandbox settings.

    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 
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
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 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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)

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) -> SandboxSession

Create a local sandbox session in a clean working directory.

Parameters:

Name Type Description Default
settings Optional[BaseSandboxSettings]

Optional settings overrides.

 None

Returns:

Type Description
SandboxSession

A local sandbox session.
Source code in src/zenml/sandboxes/local_sandbox.py 
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
def create_session(
    self, settings: Optional[BaseSandboxSettings] = None
) -> SandboxSession:
    """Create a local sandbox session in a clean working directory.

    Args:
        settings: Optional settings overrides.

    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,
    )

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 
 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
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
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 
79
80
81
@abstractmethod
def kill(self) -> None:
    """Terminates the process."""
stderr() -> Iterator[str] abstractmethod

Yields stderr lines.

Source code in src/zenml/sandboxes/process.py 
64
65
66
@abstractmethod
def stderr(self) -> Iterator[str]:
    """Yields stderr lines."""
stdout() -> Iterator[str] abstractmethod

Yields stdout lines.

Source code in src/zenml/sandboxes/process.py 
60
61
62
@abstractmethod
def stdout(self) -> Iterator[str]:
    """Yields stdout 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 
68
69
70
71
72
73
74
75
76
77
@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)

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
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
def __init__(
    self,
    *,
    id: str,
    parent: "BaseSandbox",
) -> None:
    """Initialize the sandbox session.

    Args:
        id: Session identifier.
        parent: The sandbox component that created this session.
    """
    self.id = id
    self._parent = parent
    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 
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
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 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
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 
159
160
161
162
163
164
165
166
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 
256
257
258
259
260
261
262
263
264
265
266
267
268
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 
206
207
208
209
210
211
212
213
214
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 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
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 
182
183
184
185
186
187
188
189
190
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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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) -> SandboxSession abstractmethod

Create a fresh sandbox session.

Parameters:

Name Type Description Default
settings Optional[BaseSandboxSettings]

The sandbox settings.

 None

Returns:

Type Description
SandboxSession

A new sandbox session.
Source code in src/zenml/sandboxes/base.py 
75
76
77
78
79
80
81
82
83
84
85
86
@abstractmethod
def create_session(
    self, settings: Optional[BaseSandboxSettings] = None
) -> SandboxSession:
    """Create a fresh sandbox session.

    Args:
        settings: The sandbox settings.

    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 
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
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 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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)
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) -> SandboxSession

Create a local sandbox session in a clean working directory.

Parameters:

Name Type Description Default
settings Optional[BaseSandboxSettings]

Optional settings overrides.

 None

Returns:

Type Description
SandboxSession

A local sandbox session.
Source code in src/zenml/sandboxes/local_sandbox.py 
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
def create_session(
    self, settings: Optional[BaseSandboxSettings] = None
) -> SandboxSession:
    """Create a local sandbox session in a clean working directory.

    Args:
        settings: Optional settings overrides.

    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,
    )
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)

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
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
def __init__(
    self,
    workdir: str,
    env: Dict[str, str],
    *,
    parent: "BaseSandbox",
) -> 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.
    """
    super().__init__(
        id=f"local-{uuid.uuid4().hex[:12]}",
        parent=parent,
    )
    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 
 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
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
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 
79
80
81
@abstractmethod
def kill(self) -> None:
    """Terminates the process."""
stderr() -> Iterator[str] abstractmethod

Yields stderr lines.

Source code in src/zenml/sandboxes/process.py 
64
65
66
@abstractmethod
def stderr(self) -> Iterator[str]:
    """Yields stderr lines."""
stdout() -> Iterator[str] abstractmethod

Yields stdout lines.

Source code in src/zenml/sandboxes/process.py 
60
61
62
@abstractmethod
def stdout(self) -> Iterator[str]:
    """Yields stdout 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 
68
69
70
71
72
73
74
75
76
77
@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)

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
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
def __init__(
    self,
    *,
    id: str,
    parent: "BaseSandbox",
) -> None:
    """Initialize the sandbox session.

    Args:
        id: Session identifier.
        parent: The sandbox component that created this session.
    """
    self.id = id
    self._parent = parent
    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 
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
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 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
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 
159
160
161
162
163
164
165
166
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 
256
257
258
259
260
261
262
263
264
265
266
267
268
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 
206
207
208
209
210
211
212
213
214
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 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
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 
182
183
184
185
186
187
188
189
190
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.