Skip to content

datarobot_genai.drtools.core.sandbox.base

base

Core types for the sandbox abstraction.

Defines the :class:Sandbox Protocol, the :class:SandboxResult and :class:SandboxSecurityContext data models, and the exception hierarchy (:class:SandboxError, :class:SandboxTimeout).

SandboxError

Bases: Exception

Base error raised when sandboxed execution fails.

Subclassed by :class:SandboxTimeout (timeout) and :class:SandboxInfraError (provisioning/transport failure).

Carries optional structured detail so observability can classify the failure without re-parsing the message string: exit_code (the container/process exit status, e.g. 137 for an OOM kill) and stderr (captured standard error, scanned for OOM/timeout markers).

Source code in datarobot_genai/drtools/core/sandbox/base.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class SandboxError(Exception):
    """Base error raised when sandboxed execution fails.

    Subclassed by :class:`SandboxTimeout` (timeout) and
    :class:`SandboxInfraError` (provisioning/transport failure).

    Carries optional structured detail so observability can classify the
    failure without re-parsing the message string: ``exit_code`` (the
    container/process exit status, e.g. ``137`` for an OOM kill) and
    ``stderr`` (captured standard error, scanned for OOM/timeout markers).
    """

    def __init__(
        self, message: str = "", *, exit_code: int | None = None, stderr: str = ""
    ) -> None:
        super().__init__(message)
        self.exit_code = exit_code
        self.stderr = stderr

SandboxTimeout

Bases: SandboxError

Raised when sandboxed execution exceeds the configured timeout.

Source code in datarobot_genai/drtools/core/sandbox/base.py 
52
53
class SandboxTimeout(SandboxError):  # noqa: N818  # public API name; "Timeout" intentional
    """Raised when sandboxed execution exceeds the configured timeout."""

SandboxInfraError

Bases: SandboxError

Raised when the sandbox cannot be provisioned or reached.

Distinguishes an infrastructure/transport failure before user code runs (image pull failure, workload-api unreachable) from a user-code failure, so the two are not conflated in the failure taxonomy.

Source code in datarobot_genai/drtools/core/sandbox/base.py 
56
57
58
59
60
61
62
class SandboxInfraError(SandboxError):  # noqa: N818  # mirrors SandboxError naming
    """Raised when the sandbox cannot be provisioned or reached.

    Distinguishes an infrastructure/transport failure *before* user code runs
    (image pull failure, workload-api unreachable) from a user-code failure, so
    the two are not conflated in the failure taxonomy.
    """

SandboxResult dataclass

Result of a successful sandboxed execution.

Attributes

stdout Captured standard output. stderr Captured standard error. return_value Value the user code assigned to the magic _return variable, if any. duration_s Wall-clock execution duration in seconds. exit_code Process exit code (0 for success).

Source code in datarobot_genai/drtools/core/sandbox/base.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@dataclass
class SandboxResult:
    """Result of a successful sandboxed execution.

    Attributes
    ----------
    stdout
        Captured standard output.
    stderr
        Captured standard error.
    return_value
        Value the user code assigned to the magic ``_return`` variable, if any.
    duration_s
        Wall-clock execution duration in seconds.
    exit_code
        Process exit code (``0`` for success).
    """

    stdout: str
    stderr: str
    return_value: Any
    duration_s: float
    exit_code: int

SandboxSecurityContext

Bases: BaseModel

Container security context applied to sandboxed workloads.

Mirrors the SecurityContext schema in the DataRobot workload-api (see workload_api/schemas/containers.py:169).

Defaults are deliberately the most restrictive settings that a non-admin user is allowed to apply: read-only root filesystem, drop all Linux capabilities, RuntimeDefault seccomp profile, and no privilege escalation.

Source code in datarobot_genai/drtools/core/sandbox/base.py 
 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
class SandboxSecurityContext(BaseModel):
    """Container security context applied to sandboxed workloads.

    Mirrors the ``SecurityContext`` schema in the DataRobot workload-api
    (see ``workload_api/schemas/containers.py:169``).

    Defaults are deliberately the most restrictive settings that a non-admin
    user is allowed to apply: read-only root filesystem, drop all Linux
    capabilities, ``RuntimeDefault`` seccomp profile, and no privilege
    escalation.
    """

    read_only_root_filesystem: bool = True
    capabilities_drop: list[str] = Field(default_factory=lambda: ["ALL"])
    capabilities_add: list[str] = Field(default_factory=list)
    seccomp_profile_type: SeccompProfileType = "RuntimeDefault"
    allow_privilege_escalation: bool = False

    def to_workload_api_dict(self) -> dict[str, Any]:
        """Serialize to the camelCase shape accepted by workload-api.

        Returns
        -------
        dict
            A dict with keys ``readOnlyRootFilesystem``,
            ``allowPrivilegeEscalation``, ``capabilities`` (with ``add``/``drop``
            sublists), and ``seccompProfile`` (with ``type``). Empty
            capability lists are omitted; ``capabilities`` and
            ``seccompProfile`` are omitted entirely when they would be empty.
        """
        payload: dict[str, Any] = {
            "readOnlyRootFilesystem": self.read_only_root_filesystem,
            "allowPrivilegeEscalation": self.allow_privilege_escalation,
        }

        capabilities: dict[str, list[str]] = {}
        if self.capabilities_drop:
            capabilities["drop"] = list(self.capabilities_drop)
        if self.capabilities_add:
            capabilities["add"] = list(self.capabilities_add)
        if capabilities:
            payload["capabilities"] = capabilities

        payload["seccompProfile"] = {"type": self.seccomp_profile_type}

        return payload

to_workload_api_dict 

to_workload_api_dict() -> dict[str, Any]

Serialize to the camelCase shape accepted by workload-api.

Returns

dict A dict with keys readOnlyRootFilesystem, allowPrivilegeEscalation, capabilities (with add/drop sublists), and seccompProfile (with type). Empty capability lists are omitted; capabilities and seccompProfile are omitted entirely when they would be empty.

Source code in datarobot_genai/drtools/core/sandbox/base.py 
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
def to_workload_api_dict(self) -> dict[str, Any]:
    """Serialize to the camelCase shape accepted by workload-api.

    Returns
    -------
    dict
        A dict with keys ``readOnlyRootFilesystem``,
        ``allowPrivilegeEscalation``, ``capabilities`` (with ``add``/``drop``
        sublists), and ``seccompProfile`` (with ``type``). Empty
        capability lists are omitted; ``capabilities`` and
        ``seccompProfile`` are omitted entirely when they would be empty.
    """
    payload: dict[str, Any] = {
        "readOnlyRootFilesystem": self.read_only_root_filesystem,
        "allowPrivilegeEscalation": self.allow_privilege_escalation,
    }

    capabilities: dict[str, list[str]] = {}
    if self.capabilities_drop:
        capabilities["drop"] = list(self.capabilities_drop)
    if self.capabilities_add:
        capabilities["add"] = list(self.capabilities_add)
    if capabilities:
        payload["capabilities"] = capabilities

    payload["seccompProfile"] = {"type": self.seccomp_profile_type}

    return payload

Sandbox

Bases: Protocol

Protocol for sandboxed Python code execution.

The production implementation is :class:DataRobotWorkloadSandbox (runs in the DataRobot workload-api). A local-Docker dev/test implementation is planned as a follow-up.

Source code in datarobot_genai/drtools/core/sandbox/base.py 
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
180
181
182
183
184
185
186
187
188
@runtime_checkable
class Sandbox(Protocol):
    """Protocol for sandboxed Python code execution.

    The production implementation is :class:`DataRobotWorkloadSandbox`
    (runs in the DataRobot workload-api). A local-Docker dev/test
    implementation is planned as a follow-up.
    """

    async def run(
        self,
        code: str,
        *,
        inputs: dict[str, Any] | None = None,
        externals: dict[str, Any] | None = None,
        timeout_s: float = 30.0,
    ) -> SandboxResult:
        """Execute ``code`` in an isolated environment and return the result.

        Parameters
        ----------
        code
            Python source to execute. May assign to a magic ``_return``
            variable to communicate a return value back to the caller.
        inputs
            Mapping bound as ``inputs`` in the executing namespace. Must be
            JSON-serializable for remote backends.
        externals
            Mapping bound as ``externals`` in the executing namespace for
            CodeMode-style tool injection. Most implementations do not yet
            support this and will raise ``NotImplementedError``.
        timeout_s
            Wall-clock timeout in seconds.

        Returns
        -------
        SandboxResult
            stdout/stderr capture, the ``_return`` value, duration, and
            exit code.

        Raises
        ------
        SandboxTimeout
            When execution exceeds ``timeout_s``.
        SandboxError
            When the sandbox itself fails or the user code raises.
        """
        ...

run async 

run(code: str, *, inputs: dict[str, Any] | None = None, externals: dict[str, Any] | None = None, timeout_s: float = 30.0) -> SandboxResult

Execute code in an isolated environment and return the result.

Parameters

code Python source to execute. May assign to a magic _return variable to communicate a return value back to the caller. inputs Mapping bound as inputs in the executing namespace. Must be JSON-serializable for remote backends. externals Mapping bound as externals in the executing namespace for CodeMode-style tool injection. Most implementations do not yet support this and will raise NotImplementedError. timeout_s Wall-clock timeout in seconds.

Returns

SandboxResult stdout/stderr capture, the _return value, duration, and exit code.

Raises

SandboxTimeout When execution exceeds timeout_s. SandboxError When the sandbox itself fails or the user code raises.

Source code in datarobot_genai/drtools/core/sandbox/base.py 
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
180
181
182
183
184
185
186
187
188
async def run(
    self,
    code: str,
    *,
    inputs: dict[str, Any] | None = None,
    externals: dict[str, Any] | None = None,
    timeout_s: float = 30.0,
) -> SandboxResult:
    """Execute ``code`` in an isolated environment and return the result.

    Parameters
    ----------
    code
        Python source to execute. May assign to a magic ``_return``
        variable to communicate a return value back to the caller.
    inputs
        Mapping bound as ``inputs`` in the executing namespace. Must be
        JSON-serializable for remote backends.
    externals
        Mapping bound as ``externals`` in the executing namespace for
        CodeMode-style tool injection. Most implementations do not yet
        support this and will raise ``NotImplementedError``.
    timeout_s
        Wall-clock timeout in seconds.

    Returns
    -------
    SandboxResult
        stdout/stderr capture, the ``_return`` value, duration, and
        exit code.

    Raises
    ------
    SandboxTimeout
        When execution exceeds ``timeout_s``.
    SandboxError
        When the sandbox itself fails or the user code raises.
    """
    ...