API Reference¶

def __enter__(self) -> "ArtifactSandboxSession":
    """Enter the context manager."""
    self._session.__enter__()
    return self

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit the context manager."""
    return self._session.__exit__(exc_type, exc_val, exc_tb)

def __getattr__(self, name: str) -> Any:
    """Delegate any other attributes/methods to the underlying session."""
    return getattr(self._session, name)

def clear_plots(self) -> None:
    """Manually clear all plots and reset the plot counter.

    This method can be called between runs to clear existing plots
    without creating a new session.

    Raises:
        Exception: If plot clearing is not supported or fails

    """
    if not self.enable_plotting:
        return

    self._clear_plots_in_container()

def run(
    self,
    code: str,
    libraries: list | None = None,
    timeout: float | None = None,
    clear_plots: bool = False,
) -> ExecutionResult:
    """Run code in the sandbox session and extract any generated artifacts.

    This method executes the provided code in an isolated environment and captures any
    generated artifacts (e.g., plots, figures). When plotting is enabled, it delegates
    to the language handler's run_with_artifacts method for language-specific artifact
    extraction.

    Args:
        code (str): The code to execute. Can include plotting commands from matplotlib,
                    seaborn, plotly, or other visualization libraries.
        libraries (list | None, optional): Additional libraries to install before running
                                            the code. Defaults to None.
        timeout (float | None, optional): Timeout in seconds for the code execution.
                                            Defaults to the configuration's execution_timeout
                                            (typically 60) or 60 if not configured.
        clear_plots (bool, optional): Whether to clear existing plots before running
                                        the code. Defaults to False.

    Returns:
        ExecutionResult: An object containing:
            - exit_code (int): The exit code of the execution
            - stdout (str): Standard output from the code execution
            - stderr (str): Standard error from the code execution
            - plots (list[Plot]): List of captured plots, each containing:
                - content_base64 (str): Base64 encoded plot data
                - format (PlotFormat): Format of the plot (e.g., 'png', 'svg')

    Raises:
        LanguageNotSupportPlotError: If the language does not support plot detection

    Examples:
        Basic plotting example:
        ```python
        with ArtifactSandboxSession(
            lang="python",
            verbose=True,
            image="ghcr.io/vndee/sandbox-python-311-bullseye"
        ) as session:
            code = '''
            import matplotlib.pyplot as plt
            import numpy as np

            x = np.linspace(0, 10, 100)
            y = np.sin(x)
            plt.plot(x, y)
            plt.title('Sine Wave')
            plt.show()
            '''
            result = session.run(code)
            print(f"Generated {len(result.plots)} plots")
        ```

        Multiple plot types and libraries:
        ```python
        code = '''
        import matplotlib.pyplot as plt
        import seaborn as sns
        import plotly.express as px
        import pandas as pd
        import numpy as np

        # Matplotlib plot
        plt.figure(figsize=(10, 6))
        x = np.linspace(0, 10, 100)
        plt.plot(x, np.sin(x))
        plt.title('Matplotlib: Sine Wave')
        plt.show()

        # Seaborn plot
        data = pd.DataFrame({
            'x': np.random.randn(100),
            'y': np.random.randn(100)
        })
        sns.scatterplot(data=data, x='x', y='y')
        plt.title('Seaborn: Scatter Plot')
        plt.show()

        # Plotly plot
        fig = px.line(data, x='x', y='y', title='Plotly: Line Plot')
        fig.show()
        '''

        result = session.run(code, libraries=['plotly'])

        # Save the generated plots
        for i, plot in enumerate(result.plots):
            with open(f'plot_{i}.{plot.format.value}', 'wb') as f:
                f.write(base64.b64decode(plot.content_base64))
        ```

        Installing additional libraries:
        ```python
        code = '''
        import torch
        import torch.nn as nn
        print(f"PyTorch version: {torch.__version__}")
        '''

        result = session.run(code, libraries=['torch'])
        print(result.stdout)
        ```

        Clearing plots between runs:
        ```python
        with ArtifactSandboxSession(lang="python") as session:
            # First run with plots
            plot_code = '''
            import matplotlib.pyplot as plt
            plt.plot([1, 2, 3], [1, 4, 2])
            plt.show()
            '''
            result1 = session.run(plot_code)
            print(f"First run: {len(result1.plots)} plots")

            # Clear plots and run again
            result2 = session.run("print('hello world')", clear_plots=True)
            print(f"Second run: {len(result2.plots)} plots")

            # Manual plot clearing
            session.clear_plots()
            result3 = session.run(plot_code)
            print(f"Third run: {len(result3.plots)} plots")
        ```

    """
    # Check if plotting is enabled and language supports it
    if self.enable_plotting and not self._session.language_handler.is_support_plot_detection:
        raise LanguageNotSupportPlotError(self._session.language_handler.name)

    # Clear plots if requested
    if clear_plots and self.enable_plotting:
        self._clear_plots_in_container()

    # Use config default timeout if not specified
    if timeout is not None:
        effective_timeout = timeout
    else:
        config_timeout = self._session.config.get_execution_timeout()
        effective_timeout = config_timeout if config_timeout is not None else 60

    # Delegate to language handler for language-specific artifact extraction
    result, plots = self._session.language_handler.run_with_artifacts(
        container=self._session,  # type: ignore[arg-type]
        code=code,
        libraries=libraries,
        enable_plotting=self.enable_plotting,
        output_dir="/tmp/sandbox_plots",
        timeout=int(effective_timeout),
    )

    return ExecutionResult(
        exit_code=result.exit_code,
        stdout=result.stdout,
        stderr=result.stderr,
        plots=plots,
    )

def close(self) -> None:
    """Close the interactive session."""
    self._stop_runner_process()
    if self._backend_session:
        self._backend_session.close()
    # Manually set is_open to False without calling super().close()
    # The backend session manages the timer cleanup, so we don't need to stop a duplicate timer
    self.is_open = False

def copy_from_runtime(self, src: str, dest: str) -> None:
    """Copy a file from the backend container.

    Args:
        src: Source path in the container
        dest: Destination path on the host

    """
    return self._backend_session.copy_from_runtime(src, dest)

def copy_to_runtime(self, src: str, dest: str) -> None:
    """Copy a file to the backend container.

    Args:
        src: Source path on the host
        dest: Destination path in the container

    """
    return self._backend_session.copy_to_runtime(src, dest)

def execute_command(self, command: str, workdir: str | None = None) -> ConsoleOutput:
    """Execute a command in the backend container.

    Args:
        command: The command to execute
        workdir: Optional working directory

    Returns:
        ConsoleOutput: The command output

    """
    return self._backend_session.execute_command(command, workdir=workdir)

def execute_commands(
    self, commands: list[str | tuple[str, str | None]], workdir: str | None = None
) -> ConsoleOutput:
    """Execute multiple commands in the backend container.

    Args:
        commands: List of commands to execute
        workdir: Optional working directory

    Returns:
        ConsoleOutput: The output of the last command

    """
    return self._backend_session.execute_commands(commands, workdir=workdir)

def open(self) -> None:
    """Open interactive session and prepare runtime assets.

    Raises:
        ContainerError: If runtime dependencies cannot be installed or runner fails to start

    """
    # Open backend session first (this will start the session timer)
    self._backend_session.open()
    # Sync session state from backend session without starting a duplicate timer
    # The backend session manages the timer, so we don't call super().open()
    self.is_open = True
    # Sync session start time from backend session for timeout checking
    # Access private attribute to sync timer state (backend session manages the actual timer)
    self._session_start_time = getattr(self._backend_session, "_session_start_time", None)
    # Copy container reference from backend session
    self.container = self._backend_session.container
    self.container_api = self._backend_session.container_api
    try:
        self._bootstrap_runtime()
    except Exception:
        self.close()
        raise

def run(self, code: str, libraries: list[str] | None = None, timeout: float | None = None) -> ConsoleOutput:
    """Execute code in the persistent interpreter context.

    Args:
        code: The Python code to execute
        libraries: Optional list of libraries to install before execution
        timeout: Maximum execution time in seconds

    Returns:
        ConsoleOutput containing stdout, stderr, and exit code

    Raises:
        NotOpenSessionError: If the session is not open
        ContainerError: If the interactive runtime is not ready
        SandboxTimeoutError: If execution exceeds timeout

    """
    if not self.container or not self.is_open:
        raise NotOpenSessionError

    if not self._runner_ready:
        msg = "Interactive runtime is not ready"
        raise ContainerError(msg)

    self._check_session_timeout()
    actual_timeout = timeout or self.settings.timeout or self.config.get_execution_timeout()
    self.install(libraries)

    request_id = uuid.uuid4().hex
    command_path = f"{self._commands_dir}/command-{request_id}.json"
    result_path = f"{self._results_dir}/result-{request_id}.json"

    self._write_remote_command(command_path, request_id, code)

    if not self._wait_for_remote_file(result_path, actual_timeout):
        self._interrupt_runner()
        msg = f"Interactive execution timed out after {actual_timeout} seconds"
        raise SandboxTimeoutError(msg, timeout_duration=actual_timeout)

    payload = self._read_remote_result(result_path)
    self.execute_command(f"rm -f {result_path}")
    exit_code = 0 if payload.get("success") else 1
    stdout = payload.get("stdout", "")
    stderr = payload.get("stderr", "")
    return ConsoleOutput(exit_code=exit_code, stdout=stdout, stderr=stderr)

def __post_init__(self) -> None:
    """Validate configuration values."""
    if self.history_size < 0:
        msg = "history_size must be non-negative"
        raise ValueError(msg)
    if self.poll_interval <= 0:
        msg = "poll_interval must be positive"
        raise ValueError(msg)

def success(self) -> bool:
    r"""Check if the execution was successful (exit code is 0).

    Returns:
        bool: True if `exit_code` is 0, False otherwise.

    """
    return not self.exit_code

def text(self) -> str:
    r"""Get the text representation of the console output (stdout).

    .. deprecated:: 0.1.0
        The `text` property is deprecated and will be removed in a future version.
        Use the `stdout` attribute directly instead.

    Returns:
        str: The content of the standard output stream.

    """
    warnings.warn(
        "The 'text' property is deprecated and will be removed in a future version. "
        "Use 'stdout' attribute directly instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return self.stdout

def to_json(self, include_plots: bool = False) -> str:
    r"""Get the JSON representation of the execution result.

    Args:
        include_plots (bool): Whether to include the plots in the JSON representation.

    Returns:
        str: The JSON representation of the execution result.

    """
    result = self.__dict__.copy()
    if not include_plots and "plots" in result:
        result.pop("plots", None)

    return json.dumps(result, indent=2)