utils.sandbox

sandbox

Functions:

Name	Description
build_docker_image	Builds the Docker image.
cleanup_docker_artifacts	Cleans up the generated files and Docker image.
run_in_docker	Runs the code in a Docker container with optional mounts.

Classes

Functions:

build_docker_image

build_docker_image(where: str)

Builds the Docker image.

Source code in unaiverse/utils/sandbox.py

def build_docker_image(where: str):
    """Builds the Docker image."""
    print(f"Building Docker image '{DOCKER_IMAGE_NAME}'...")

    try:

        # The '.' at the end means build from the current directory
        subprocess.run(["docker", "build", "-t", DOCKER_IMAGE_NAME, where], check=True)
        print(f"Docker image '{DOCKER_IMAGE_NAME}' built successfully.")
        return True
    except subprocess.CalledProcessError as e:
        print(f"Error building Docker image: {e}")
        return False

cleanup_docker_artifacts

cleanup_docker_artifacts(where: str)

Cleans up the generated files and Docker image.

Source code in unaiverse/utils/sandbox.py

def cleanup_docker_artifacts(where: str):
    """Cleans up the generated files and Docker image."""
    print("Cleaning...")

    # Stop and remove container if it's still running (e.g., if previous run failed)
    try:
        print(f"Attempting to stop and remove container '{CONTAINER_NAME}' (if running)...")
        subprocess.run(["docker", "stop", CONTAINER_NAME],
                       check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        subprocess.run(["docker", "rm", CONTAINER_NAME],
                       check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    except Exception as e:
        print(f"Error during preliminary container cleanup: {e}")

    # Remove the Docker image
    try:
        print(f"Removing Docker image '{DOCKER_IMAGE_NAME}'...")
        subprocess.run(["docker", "rmi", DOCKER_IMAGE_NAME], check=True)
        print("Docker image removed.")
    except subprocess.CalledProcessError as e:
        print(f"Error removing Docker image '{DOCKER_IMAGE_NAME}': {e}")

    # Remove the generated Dockerfile
    if os.path.exists(os.path.join(where, "Dockerfile")):
        os.remove(os.path.join(where, "Dockerfile"))
        print("Removed Dockerfile.")

run_in_docker

run_in_docker(file_to_run: str, read_only_host_paths: list[str] = None, writable_host_paths: list[str] = None)

Runs the code in a Docker container with optional mounts.

Source code in unaiverse/utils/sandbox.py

def run_in_docker(file_to_run: str, read_only_host_paths: list[str] = None, writable_host_paths: list[str] = None):
    """Runs the code in a Docker container with optional mounts."""
    print(f"\nRunning code in Docker container '{CONTAINER_NAME}'...")

    # Building command (it will continue below...)
    command = ["docker", "run",
               "--rm",  # Automatically remove the container when it exits
               "-e", "PYTHONUNBUFFERED=1",  # Ensure Python output is unbuffered
               "-e", "NODE_STARTING_PORT",
               "--name", CONTAINER_NAME]

    if sys.platform.startswith('linux'):

        # Linux
        command.extend(["--net", "host"]),  # Expose the host network (in macOS and Windows it is still a virtual host)
    else:

        # Not-linux: check ports (adding -p port:port)
        port_int = int(os.getenv("NODE_STARTING_PORT", "0"))
        if port_int > 0:
            command.extend(["-p", str(port_int + 0) + ":" + str(port_int + 0)])
            command.extend(["-p", str(port_int + 1) + ":" + str(port_int + 1) + "/udp"])
            command.extend(["-p", str(port_int + 2) + ":" + str(port_int + 2)])
            command.extend(["-p", str(port_int + 3) + ":" + str(port_int + 3) + "/udp"])

    # Add read-only mount if path is provided
    if read_only_host_paths is not None and len(read_only_host_paths) > 0:
        for path in read_only_host_paths:

            # Ensure the host path exists and is a directory
            if not os.path.isdir(path):
                print(
                    f"Error: Read-only host path '{path}' does not exist or is not a directory. Cannot mount.")
                return False
            else:

                # Augmenting command
                path = os.path.abspath(path)
                command.extend(["-v", f"{path}:{path}:ro"])
                print(f"Mounted host '{path}' as read-only to container")

    # Add writable mount if path is provided
    if writable_host_paths is not None and len(writable_host_paths) > 0:
        for path in writable_host_paths:

            # Ensure the host path exists and is a directory
            if not os.path.isdir(path):
                print(
                    f"Error: Writable host path '{path}' does not exist or is not a directory. Cannot mount.")
                return False
            else:

                # Augmenting command
                path = os.path.abspath(path)
                command.extend(["-v", f"{path}:{path}"])
                print(f"Mounted host '{path}' as writable to container")

    # Completing command
    command.append(DOCKER_IMAGE_NAME)

    try:

        # Running the prepared command... (using Popen to stream output in real-time)
        try:
            command.extend(["python3", file_to_run])
            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
            for line in iter(process.stdout.readline, ''):
                sys.stdout.write(line)
            process.wait()  # Wait for the process to finish
            if process.returncode != 0:
                print(f"Container exited with non-zero status code: {process.returncode}")
        except KeyboardInterrupt:
            pass

        print(f"\nContainer '{CONTAINER_NAME}' finished execution.")
        return True
    except FileNotFoundError:
        print("Error: Docker command not found. Is Docker installed and in your PATH?")
        print("Please ensure Docker is installed and running.")
        return False
    except subprocess.CalledProcessError as e:
        print(f"Error running Docker container: {e}")
        return False