Skip to content

orchestr8.sandbox_client

A wrapper around docker client for building and running containers.

client = SandboxClient()

# Check if image build exists locally
client.image_exists("ubuntu:22.04", where="local")

# Check if image is available in registry
client.image_exists("ubuntu:22.04", where="registry")

# Pull image from Docker Hub
client.build_image("ubuntu:22.04")

# Build image from Dockerfile
client.build_image("my-image:my-tag", dockerfile=Path("/path/to/Dockerfile"))

# Run container
container = client.run_container("ubuntu:22.04", detach=True)
Source code in orchestr8/sandbox_client.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 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
 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
class SandboxClient(Logger):
    """
    A wrapper around docker client for building and running containers.

    ```python
    client = SandboxClient()

    # Check if image build exists locally
    client.image_exists("ubuntu:22.04", where="local")

    # Check if image is available in registry
    client.image_exists("ubuntu:22.04", where="registry")

    # Pull image from Docker Hub
    client.build_image("ubuntu:22.04")

    # Build image from Dockerfile
    client.build_image("my-image:my-tag", dockerfile=Path("/path/to/Dockerfile"))

    # Run container
    container = client.run_container("ubuntu:22.04", detach=True)
    ```
    """

    def __init__(self, **config: Any) -> None:
        """
        Initialize the client.

        Args:
            config: Docker client configuration
        """
        from docker.errors import DockerException  # type: ignore[import-untyped]

        try:
            import docker  # type: ignore[import-untyped]

            self._instance = docker.DockerClient(**config) if config else docker.from_env()
        except DockerException as e:
            if "The system cannot find the file specified." in str(e):
                raise DockerException("Unable to connect. Make sure docker service is running.") from e
            raise

    def image_exists(self, image: str, *, where: Literal["local", "registry"] = "local") -> bool:
        """
        Check if an image exists.

        Args:
            image: The image name
            where: The source to check. Defaults to "local"

        Returns:
            True if the image exists, False otherwise
        """
        from docker.errors import NotFound

        try:
            if where == "local":  # noqa: SIM108
                _ = self._instance.images.get(image)
            else:
                _ = self._instance.images.get_registry_data(image)
            return True  # noqa: TRY300
        except NotFound:
            return False

    def build_image(self, image: str, *, dockerfile: Path | BytesIO | None = None) -> None:
        """
        Build an image from pulling it from the registry or building it from a dockerfile.

        Args:
            image: The image name
            dockerfile: The dockerfile to build. Defaults to None

        Raises:
            Exception: If the image not found in the registry
        """
        if self.image_exists(image, where="local"):
            self.logger.info(f"Skipping build process. Image {image!r} already exists.")
            return

        if not dockerfile:
            if not self.image_exists(image, where="registry"):
                raise Exception(f"Image {image!r} not found on docker hub.")

            self.logger.info(f"Pulling image {image!r} from registry")
            _ = self._instance.images.pull(*(s if len(s := image.split(":")) == 2 else s[:1]))
            return

        if isinstance(dockerfile, BytesIO):  # noqa: SIM108
            fileobj = dockerfile
        else:
            fileobj = BytesIO(dockerfile.read_bytes())

        self.logger.info(f"Building image {image!r} from dockerfile")
        for line in self._instance.api.build(fileobj=fileobj, tag=image, rm=True, quiet=False, decode=True):
            if (log := line.get("stream")) is not None:
                self.logger.info(log.strip())

    def run_container(  # type: ignore[no-any-unimported]
        self, image: str, *, volumes: Dict[Path, VolumeValue] | None = None, **run_kwargs: Any
    ) -> Container:
        """
        Run a container from a locally available image build.
        Use `.build_image` before requesting for container instance.

        Args:
            image: The image to create container instance from
            volumes: Volumes to mount inside the container. Defaults to None
            run_kwargs: Keyword arguments to pass to `docker run`

        Returns:
            The container instance

        Raises:
            Exception: If the image build is not found
        """

        if not self.image_exists(image, where="local"):
            raise Exception(f"Build not found for image {image!r}.")

        if volumes:
            run_kwargs["volumes"] = {}
            for path, vv in volumes.items():
                run_kwargs["volumes"][str(path)] = vv

        self.logger.info(f"Starting up container from image {image!r}")
        return self._instance.containers.run(image=image, **run_kwargs)

    def copy_path_to_container(self, __id: str, *, src: str | Path, target: str | Path) -> None:
        """
        Copy a path from the local system to the container.

        Args:
            __id: Container ID
            src: Source path in the local system
            target: Target path inside the container.

        Raises:
            FileNotFoundError: If the source path does not exist
        """
        path = Path(src)
        if not path.exists():
            raise FileNotFoundError(f"Source path not found: {path!s}")

        self.logger.info(f"Copying {str(path)!r} to {str(target)!r} inside container(sid={__id[:12]!r})")
        with BytesIO() as buffer, tar_open(fileobj=buffer, mode="w") as tar:
            tar.add(path, arcname=path.name)
            buffer.seek(0)

            self._instance.api.put_archive(__id, path=Path(target).as_posix(), data=buffer)

build_image

Build an image from pulling it from the registry or building it from a dockerfile.

Parameters:

Name Type Description Default
image str

The image name

 required
dockerfile Path | BytesIO | None

The dockerfile to build. Defaults to None

 None

Raises:

Type Description
Exception

If the image not found in the registry
Source code in orchestr8/sandbox_client.py 
 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
def build_image(self, image: str, *, dockerfile: Path | BytesIO | None = None) -> None:
    """
    Build an image from pulling it from the registry or building it from a dockerfile.

    Args:
        image: The image name
        dockerfile: The dockerfile to build. Defaults to None

    Raises:
        Exception: If the image not found in the registry
    """
    if self.image_exists(image, where="local"):
        self.logger.info(f"Skipping build process. Image {image!r} already exists.")
        return

    if not dockerfile:
        if not self.image_exists(image, where="registry"):
            raise Exception(f"Image {image!r} not found on docker hub.")

        self.logger.info(f"Pulling image {image!r} from registry")
        _ = self._instance.images.pull(*(s if len(s := image.split(":")) == 2 else s[:1]))
        return

    if isinstance(dockerfile, BytesIO):  # noqa: SIM108
        fileobj = dockerfile
    else:
        fileobj = BytesIO(dockerfile.read_bytes())

    self.logger.info(f"Building image {image!r} from dockerfile")
    for line in self._instance.api.build(fileobj=fileobj, tag=image, rm=True, quiet=False, decode=True):
        if (log := line.get("stream")) is not None:
            self.logger.info(log.strip())

copy_path_to_container

Copy a path from the local system to the container.

Parameters:

Name Type Description Default
__id str

Container ID

 required
src str | Path

Source path in the local system

 required
target str | Path

Target path inside the container.

 required

Raises:

Type Description
FileNotFoundError

If the source path does not exist
Source code in orchestr8/sandbox_client.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def copy_path_to_container(self, __id: str, *, src: str | Path, target: str | Path) -> None:
    """
    Copy a path from the local system to the container.

    Args:
        __id: Container ID
        src: Source path in the local system
        target: Target path inside the container.

    Raises:
        FileNotFoundError: If the source path does not exist
    """
    path = Path(src)
    if not path.exists():
        raise FileNotFoundError(f"Source path not found: {path!s}")

    self.logger.info(f"Copying {str(path)!r} to {str(target)!r} inside container(sid={__id[:12]!r})")
    with BytesIO() as buffer, tar_open(fileobj=buffer, mode="w") as tar:
        tar.add(path, arcname=path.name)
        buffer.seek(0)

        self._instance.api.put_archive(__id, path=Path(target).as_posix(), data=buffer)

image_exists

Check if an image exists.

Parameters:

Name Type Description Default
image str

The image name

 required
where Literal['local', 'registry']

The source to check. Defaults to "local"

 'local'

Returns:

Type Description
bool

True if the image exists, False otherwise
Source code in orchestr8/sandbox_client.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def image_exists(self, image: str, *, where: Literal["local", "registry"] = "local") -> bool:
    """
    Check if an image exists.

    Args:
        image: The image name
        where: The source to check. Defaults to "local"

    Returns:
        True if the image exists, False otherwise
    """
    from docker.errors import NotFound

    try:
        if where == "local":  # noqa: SIM108
            _ = self._instance.images.get(image)
        else:
            _ = self._instance.images.get_registry_data(image)
        return True  # noqa: TRY300
    except NotFound:
        return False

run_container

Run a container from a locally available image build. Use .build_image before requesting for container instance.

Parameters:

Name Type Description Default
image str

The image to create container instance from

 required
volumes Dict[Path, VolumeValue] | None

Volumes to mount inside the container. Defaults to None

 None
run_kwargs Any

Keyword arguments to pass to docker run

 {}

Returns:

Type Description
Container

The container instance

Raises:

Type Description
Exception

If the image build is not found
Source code in orchestr8/sandbox_client.py 
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
def run_container(  # type: ignore[no-any-unimported]
    self, image: str, *, volumes: Dict[Path, VolumeValue] | None = None, **run_kwargs: Any
) -> Container:
    """
    Run a container from a locally available image build.
    Use `.build_image` before requesting for container instance.

    Args:
        image: The image to create container instance from
        volumes: Volumes to mount inside the container. Defaults to None
        run_kwargs: Keyword arguments to pass to `docker run`

    Returns:
        The container instance

    Raises:
        Exception: If the image build is not found
    """

    if not self.image_exists(image, where="local"):
        raise Exception(f"Build not found for image {image!r}.")

    if volumes:
        run_kwargs["volumes"] = {}
        for path, vv in volumes.items():
            run_kwargs["volumes"][str(path)] = vv

    self.logger.info(f"Starting up container from image {image!r}")
    return self._instance.containers.run(image=image, **run_kwargs)