DockerCommandLineCodeExecutor

def __init__(
    self,
    image: str = "python:3-slim",
    container_name: Optional[str] = None,
    timeout: int = 60,
    work_dir: Optional[Union[Path, str]] = None,
    bind_dir: Optional[Union[Path, str]] = None,
    auto_remove: bool = True,
    stop_container: bool = True,
    execution_policies: Optional[dict[str, bool]] = None,
):
    """(Experimental) A code executor class that executes code through
    a command line environment in a Docker container.

    The executor first saves each code block in a file in the working
    directory, and then executes the code file in the container.
    The executor executes the code blocks in the order they are received.
    Currently, the executor only supports Python and shell scripts.
    For Python code, use the language "python" for the code block.
    For shell scripts, use the language "bash", "shell", or "sh" for the code
    block.

    Args:
        image: Docker image to use for code execution. Defaults to "python:3-slim".
        container_name: Name of the Docker container which is created. If None, will autogenerate a name. Defaults to None.
        timeout: The timeout for code execution. Defaults to 60.
        work_dir: The working directory for the code execution. Defaults to Path(".").
        bind_dir: The directory that will be bound to the code executor container. Useful for cases where you want to spawn
            the container from within a container. Defaults to work_dir.
        auto_remove: If true, will automatically remove the Docker container when it is stopped. Defaults to True.
        stop_container: If true, will automatically stop the
            container when stop is called, when the context manager exits or when
            the Python process exits with atext. Defaults to True.
        execution_policies: A dictionary mapping language names to boolean values that determine
            whether code in that language should be executed. True means code in that language
            will be executed, False means it will only be saved to a file. This overrides the
            default execution policies. Defaults to None.

    Raises:
        ValueError: On argument error, or if the container fails to start.
    """
    work_dir = work_dir if work_dir is not None else Path()

    if timeout < 1:
        raise ValueError("Timeout must be greater than or equal to 1.")

    if isinstance(work_dir, str):
        work_dir = Path(work_dir)
    work_dir.mkdir(exist_ok=True)

    if bind_dir is None:
        bind_dir = work_dir
    elif isinstance(bind_dir, str):
        bind_dir = Path(bind_dir)

    client = docker.from_env()
    # Check if the image exists
    try:
        client.images.get(image)
    except ImageNotFound:
        logging.info(f"Pulling image {image}...")
        # Let the docker exception escape if this fails.
        client.images.pull(image)

    if container_name is None:
        container_name = f"autogen-code-exec-{uuid.uuid4()}"

    # Start a container from the image, read to exec commands later
    self._container = client.containers.create(
        image,
        name=container_name,
        entrypoint="/bin/sh",
        tty=True,
        auto_remove=auto_remove,
        volumes={str(bind_dir.resolve()): {"bind": "/workspace", "mode": "rw"}},
        working_dir="/workspace",
    )
    self._container.start()

    _wait_for_ready(self._container)

    def cleanup() -> None:
        try:
            container = client.containers.get(container_name)
            container.stop()
        except docker.errors.NotFound:
            pass
        atexit.unregister(cleanup)

    if stop_container:
        atexit.register(cleanup)

    self._cleanup = cleanup

    # Check if the container is running
    if self._container.status != "running":
        raise ValueError(f"Failed to start container from image {image}. Logs: {self._container.logs()}")

    self._timeout = timeout
    self._work_dir: Path = work_dir
    self._bind_dir: Path = bind_dir
    self.execution_policies = self.DEFAULT_EXECUTION_POLICY.copy()
    if execution_policies is not None:
        self.execution_policies.update(execution_policies)