Skip to content

docker_wrapper API Documentation

DockerImage

Class providing the necessary interface to interact with a Docker image and create containers.

Source code in src/docker_wrapper/docker_helpers.py 
 14
 15
 16
 17
 18
 19
 20
 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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
class DockerImage:
    """Class providing the necessary interface to interact with a Docker image
    and create containers.
    """

    NAME = "UNDEFINED"

    def __init__(self, docker_registry_prefix: str = "", **kwargs: int) -> None:
        """Initialize the DockerImage object.

        Args:
            docker_registry_prefix (str, optional): The prefix of the Docker registry URL.
                Defaults to "".
            **kwargs: Additional keyword arguments.
        """
        self.docker_client = docker.from_env()
        self.name = "UNDEFINED"
        self.docker_folder = ""
        self.version = ""
        self.repo_url = docker_registry_prefix or None

    @staticmethod
    def _exec_cmd(cmd: List[str]) -> None:
        """Helper function to execute a shell command, and log it it as well

        Args:
            cmd (List[str]): The command and its arguments in a list format.
        """
        logging.info(" ".join(cmd))
        subprocess.check_call(cmd, stdout=sys.stdout, stderr=sys.stderr)

    @staticmethod
    def folder_hash(docker_path: str) -> str:
        """Compute the hash  of the docker image based on the
        contents of the Docker folder of the project

        Args:
            docker_path (str): Path of the Docker folder that contains the Dockerfile
                the entrypoint and any other related logic.

        Returns:
            str: Return the hash of the contents of the Docker folder.
        """
        return checksumdir.dirhash(docker_path, "sha1")

    @property
    def image_hash(self) -> str:
        """Return the full hash of the image

        Returns:
            str: Hash value.
        """
        return self.folder_hash(self.docker_folder)

    @property
    def image_tag(self) -> str:
        """Return the tag of the image

        If the image has an explicit numeric version return that, else
        return the first 10 characters of the image hash.

        Returns:
            str: Return string to be used at the image tag.
        """
        if self.version:
            return self.version
        return self.image_hash[:10]

    @property
    def tagged_name(self) -> str:
        """Return the tuple <IMAGE_NAME>:<IMAGE_TAG>

        Returns:
            str: String result
        """
        return f"{self.name}:{self.image_tag}"

    @property
    def image_url(self) -> str:
        """Return the full image URL, that is the
            <REPO_URL>/<IMAGE_NAME>:<IMAGE_TAG>

        Returns:
            str: String result
        """
        if not self.repo_url:
            return self.tagged_name
        return f"{self.repo_url}/{self.tagged_name}"

    def build_image(self, force_build: bool = False) -> None:
        """Build the Docker image if a specific version does not
        already exist.

        Args:
            force_build (bool, optional): Iff True then build the image regardless
                . Defaults to False.
        """
        image_url = self.image_url
        if self.image_exists(image_url) and not force_build:
            logging.info(f"Image: {image_url} already exists, not rebuilding")
            return
        cmd = ["docker", "build", self.docker_folder, "-t", image_url]
        self._exec_cmd(cmd)

    def pull(self) -> None:
        """Pull the image from the registry."""
        cmd = ["docker", "pull", self.image_url]
        self._exec_cmd(cmd)

    def push(self) -> None:
        """Push the image to the registry"""
        cmd = ["docker", "push", self.image_url]
        self._exec_cmd(cmd)

    def image_exists(self, url: str) -> bool:
        """Return true if a docker image exists locally.

        Args:
            url (str): Full URL:TAG name of the image

        Returns:
            bool: True iff the image exists locally
        """
        try:
            self.docker_client.images.get(url)
            return True
        except docker.errors.ImageNotFound:
            return False

    def get_docker_run_args(self) -> List[str]:
        """Return the list of additional arguments to pass to the docker run command

        This function should be overridden by the child classes to provide
        the necessary arguments.
        """
        return []

    def run(
        self,
        project_dir: Path,
        prompt: bool = False,
        mount_home: bool = False,
        cmds: Optional[List[str]] = None,
        network: Optional[str] = None,
        privileged: bool = False,
        enable_gui: bool = False,
        ports: Optional[List[str]] = None,
        volumes: Optional[List[str]] = None,
        envs: Optional[List[str]] = None,
        enable_sudo: bool = False,
        mount_host_passwd: bool = True,
    ) -> None:
        """Run a container from the Docker Image

        Args:
            project_dir (Path): Project we want to work inside the container.
                Mount the volume inside the container
            prompt (bool, optional): Iff true start the container in interactive mode and
                start a prompt to provide to the user. Defaults to False.
            cmds (Optional[List[str]], optional): List of commands to run inside the container.
                Defaults to None.
            network (Optional[str], optional): Name of the network to connect the container to.
                Defaults to None.
            privileged (bool, optional): Enable privileged mode. Defaults to False.
            enable_gui (bool, optional): Enable support for starting GUI apps inside the container.
                Defaults to False.
            ports (Optional[List[str]], optional): List of ports to enable to open from the
                container. Defaults to None.
            volumes (Optional[List[str]], optional): List of volumes to mount inside the container.
            enable_sudo (bool, optional): Enable sudo inside the container. Defaults to False. This
                option mounts the sudoers file inside the container.
            mount_host_passwd (bool, optional): Mount the host passwd related files inside the
                container and make use of the `-u` Docker option to map the user to the host user.
        """
        if not (bool(prompt) ^ bool(cmds)):
            raise RuntimeError("Either or neither prompt or cmds are set, only one could be set")
        image_url = self.image_url
        logging.debug(f"Using Image: {image_url}")
        if not self.image_exists(image_url):
            logging.debug(f"Image {image_url} does not exist")
            self.build_image()
        uid = os.getuid()
        gid = os.getgid()
        username = getpass.getuser()
        logging.debug("uid:{}, gid:{}, username:{}".format(uid, gid, username))
        home_dir = os.path.expanduser("~")
        cmd = ["docker"]
        cmd += ["run", "--rm", "--hostname=Docker"]
        project_realpath = os.path.realpath(project_dir)
        if privileged:
            cmd += ["--privileged"]
        if enable_gui:
            cmd += [
                "-e",
                "DISPLAY=$DISPLAY",
                "-v",
                "/tmp/.X11-unix:/tmp/.X11-unix",
            ]
        if volumes:
            for v in volumes:
                cmd += ["-v", v]
        if network:
            cmd += [f"--network={network}"]
        if mount_host_passwd:
            cmd += [
                "-v",
                "/etc/group:/etc/group:ro",
                "-v",
                "/etc/passwd:/etc/passwd:ro",
                "-v",
                "/etc/shadow:/etc/shadow:ro",
                "-u",
                f"{uid}:{gid}",
            ]

            # Enable sudo only if we are mounting host password files
            if enable_sudo:
                cmd += ["-v", "/etc/sudoers.d:/etc/sudoers.d:ro"]

        cmd += [
            "-e",
            f"SRC_DIR={project_realpath}",
        ]
        if ports:
            for port in ports:
                cmd += ["-p", f"{port}:{port}"]

        if prompt:
            cmd += ["-t", "-i", "-e", "PROMPT=1"]

        cmd += self.get_docker_run_args()
        if mount_home:
            cmd += [
                "-v",
                home_dir + ":" + home_dir,
            ]
        if envs:
            for env in envs:
                cmd += ["-e", env]
        cmd += [
            "-v",
            f"{project_realpath}:{project_realpath}",
            image_url,
        ]

        if cmds:
            cmd.append(" ".join(cmds))

        self._exec_cmd(cmd)

image_hash: str property

Return the full hash of the image

Returns:

Name Type Description
str str

Hash value.

image_tag: str property

Return the tag of the image

If the image has an explicit numeric version return that, else return the first 10 characters of the image hash.

Returns:

Name Type Description
str str

Return string to be used at the image tag.

image_url: str property

Return the full image URL, that is the /:

Returns:

Name Type Description
str str

String result

tagged_name: str property

Return the tuple :

Returns:

Name Type Description
str str

String result

__init__(docker_registry_prefix='', **kwargs)

Initialize the DockerImage object.

Parameters:

Name Type Description Default
docker_registry_prefix str

The prefix of the Docker registry URL. Defaults to "".

 ''
**kwargs int

Additional keyword arguments.

 {}
Source code in src/docker_wrapper/docker_helpers.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(self, docker_registry_prefix: str = "", **kwargs: int) -> None:
    """Initialize the DockerImage object.

    Args:
        docker_registry_prefix (str, optional): The prefix of the Docker registry URL.
            Defaults to "".
        **kwargs: Additional keyword arguments.
    """
    self.docker_client = docker.from_env()
    self.name = "UNDEFINED"
    self.docker_folder = ""
    self.version = ""
    self.repo_url = docker_registry_prefix or None

build_image(force_build=False)

Build the Docker image if a specific version does not already exist.

Parameters:

Name Type Description Default
force_build bool

Iff True then build the image regardless . Defaults to False.

 False
Source code in src/docker_wrapper/docker_helpers.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def build_image(self, force_build: bool = False) -> None:
    """Build the Docker image if a specific version does not
    already exist.

    Args:
        force_build (bool, optional): Iff True then build the image regardless
            . Defaults to False.
    """
    image_url = self.image_url
    if self.image_exists(image_url) and not force_build:
        logging.info(f"Image: {image_url} already exists, not rebuilding")
        return
    cmd = ["docker", "build", self.docker_folder, "-t", image_url]
    self._exec_cmd(cmd)

folder_hash(docker_path) staticmethod

Compute the hash of the docker image based on the contents of the Docker folder of the project

Parameters:

Name Type Description Default
docker_path str

Path of the Docker folder that contains the Dockerfile the entrypoint and any other related logic.

 required

Returns:

Name Type Description
str str

Return the hash of the contents of the Docker folder.
Source code in src/docker_wrapper/docker_helpers.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
@staticmethod
def folder_hash(docker_path: str) -> str:
    """Compute the hash  of the docker image based on the
    contents of the Docker folder of the project

    Args:
        docker_path (str): Path of the Docker folder that contains the Dockerfile
            the entrypoint and any other related logic.

    Returns:
        str: Return the hash of the contents of the Docker folder.
    """
    return checksumdir.dirhash(docker_path, "sha1")

get_docker_run_args()

Return the list of additional arguments to pass to the docker run command

This function should be overridden by the child classes to provide the necessary arguments.

Source code in src/docker_wrapper/docker_helpers.py 
143
144
145
146
147
148
149
def get_docker_run_args(self) -> List[str]:
    """Return the list of additional arguments to pass to the docker run command

    This function should be overridden by the child classes to provide
    the necessary arguments.
    """
    return []

image_exists(url)

Return true if a docker image exists locally.

Parameters:

Name Type Description Default
url str

Full URL:TAG name of the image

 required

Returns:

Name Type Description
bool bool

True iff the image exists locally
Source code in src/docker_wrapper/docker_helpers.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def image_exists(self, url: str) -> bool:
    """Return true if a docker image exists locally.

    Args:
        url (str): Full URL:TAG name of the image

    Returns:
        bool: True iff the image exists locally
    """
    try:
        self.docker_client.images.get(url)
        return True
    except docker.errors.ImageNotFound:
        return False

pull()

Pull the image from the registry.

Source code in src/docker_wrapper/docker_helpers.py 
118
119
120
121
def pull(self) -> None:
    """Pull the image from the registry."""
    cmd = ["docker", "pull", self.image_url]
    self._exec_cmd(cmd)

push()

Push the image to the registry

Source code in src/docker_wrapper/docker_helpers.py 
123
124
125
126
def push(self) -> None:
    """Push the image to the registry"""
    cmd = ["docker", "push", self.image_url]
    self._exec_cmd(cmd)

run(project_dir, prompt=False, mount_home=False, cmds=None, network=None, privileged=False, enable_gui=False, ports=None, volumes=None, envs=None, enable_sudo=False, mount_host_passwd=True)

Run a container from the Docker Image

Parameters:

Name Type Description Default
project_dir Path

Project we want to work inside the container. Mount the volume inside the container

 required
prompt bool

Iff true start the container in interactive mode and start a prompt to provide to the user. Defaults to False.

 False
cmds Optional[List[str]]

List of commands to run inside the container. Defaults to None.

 None
network Optional[str]

Name of the network to connect the container to. Defaults to None.

 None
privileged bool

Enable privileged mode. Defaults to False.

 False
enable_gui bool

Enable support for starting GUI apps inside the container. Defaults to False.

 False
ports Optional[List[str]]

List of ports to enable to open from the container. Defaults to None.

 None
volumes Optional[List[str]]

List of volumes to mount inside the container.

 None
enable_sudo bool

Enable sudo inside the container. Defaults to False. This option mounts the sudoers file inside the container.

 False
mount_host_passwd bool

Mount the host passwd related files inside the container and make use of the -u Docker option to map the user to the host user.

 True
Source code in src/docker_wrapper/docker_helpers.py 
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
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def run(
    self,
    project_dir: Path,
    prompt: bool = False,
    mount_home: bool = False,
    cmds: Optional[List[str]] = None,
    network: Optional[str] = None,
    privileged: bool = False,
    enable_gui: bool = False,
    ports: Optional[List[str]] = None,
    volumes: Optional[List[str]] = None,
    envs: Optional[List[str]] = None,
    enable_sudo: bool = False,
    mount_host_passwd: bool = True,
) -> None:
    """Run a container from the Docker Image

    Args:
        project_dir (Path): Project we want to work inside the container.
            Mount the volume inside the container
        prompt (bool, optional): Iff true start the container in interactive mode and
            start a prompt to provide to the user. Defaults to False.
        cmds (Optional[List[str]], optional): List of commands to run inside the container.
            Defaults to None.
        network (Optional[str], optional): Name of the network to connect the container to.
            Defaults to None.
        privileged (bool, optional): Enable privileged mode. Defaults to False.
        enable_gui (bool, optional): Enable support for starting GUI apps inside the container.
            Defaults to False.
        ports (Optional[List[str]], optional): List of ports to enable to open from the
            container. Defaults to None.
        volumes (Optional[List[str]], optional): List of volumes to mount inside the container.
        enable_sudo (bool, optional): Enable sudo inside the container. Defaults to False. This
            option mounts the sudoers file inside the container.
        mount_host_passwd (bool, optional): Mount the host passwd related files inside the
            container and make use of the `-u` Docker option to map the user to the host user.
    """
    if not (bool(prompt) ^ bool(cmds)):
        raise RuntimeError("Either or neither prompt or cmds are set, only one could be set")
    image_url = self.image_url
    logging.debug(f"Using Image: {image_url}")
    if not self.image_exists(image_url):
        logging.debug(f"Image {image_url} does not exist")
        self.build_image()
    uid = os.getuid()
    gid = os.getgid()
    username = getpass.getuser()
    logging.debug("uid:{}, gid:{}, username:{}".format(uid, gid, username))
    home_dir = os.path.expanduser("~")
    cmd = ["docker"]
    cmd += ["run", "--rm", "--hostname=Docker"]
    project_realpath = os.path.realpath(project_dir)
    if privileged:
        cmd += ["--privileged"]
    if enable_gui:
        cmd += [
            "-e",
            "DISPLAY=$DISPLAY",
            "-v",
            "/tmp/.X11-unix:/tmp/.X11-unix",
        ]
    if volumes:
        for v in volumes:
            cmd += ["-v", v]
    if network:
        cmd += [f"--network={network}"]
    if mount_host_passwd:
        cmd += [
            "-v",
            "/etc/group:/etc/group:ro",
            "-v",
            "/etc/passwd:/etc/passwd:ro",
            "-v",
            "/etc/shadow:/etc/shadow:ro",
            "-u",
            f"{uid}:{gid}",
        ]

        # Enable sudo only if we are mounting host password files
        if enable_sudo:
            cmd += ["-v", "/etc/sudoers.d:/etc/sudoers.d:ro"]

    cmd += [
        "-e",
        f"SRC_DIR={project_realpath}",
    ]
    if ports:
        for port in ports:
            cmd += ["-p", f"{port}:{port}"]

    if prompt:
        cmd += ["-t", "-i", "-e", "PROMPT=1"]

    cmd += self.get_docker_run_args()
    if mount_home:
        cmd += [
            "-v",
            home_dir + ":" + home_dir,
        ]
    if envs:
        for env in envs:
            cmd += ["-e", env]
    cmd += [
        "-v",
        f"{project_realpath}:{project_realpath}",
        image_url,
    ]

    if cmds:
        cmd.append(" ".join(cmds))

    self._exec_cmd(cmd)

Command line interface of the Docker Wrapper module

LoggingLevel

Bases: str, Enum

Enum holding the different logging argument values.

Attributes:

Name Type Description
CRITICAL str

Represents the 'CRITICAL' logging level.
ERROR str

Represents the 'ERROR' logging level.
WARNING str

Represents the 'WARNING' logging level.
INFO str

Represents the 'INFO' logging level.
DEBUG str

Represents the 'DEBUG' logging level.
Source code in src/docker_wrapper/cli.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class LoggingLevel(str, Enum):
    """
    Enum holding the different logging argument values.

    Attributes:
        CRITICAL (str): Represents the 'CRITICAL' logging level.
        ERROR (str): Represents the 'ERROR' logging level.
        WARNING (str): Represents the 'WARNING' logging level.
        INFO (str): Represents the 'INFO' logging level.
        DEBUG (str): Represents the 'DEBUG' logging level.
    """

    CRITICAL = "CRITICAL"
    ERROR = "ERROR"
    WARNING = "WARNING"
    INFO = "INFO"
    DEBUG = "DEBUG"

__create_image(image_name, **kwargs)

Instantiate an DockerImage object (or its subclass) for the specified Image

Parameters:

Name Type Description Default
image_name str

Name of the image to create an image for.

 required
**kwargs Dict[str, str]

Additional arguments to pass to the DockerImage object.

 {}

Raises:

Type Description
Exit

Failure if the image is not found

Returns:

Type Description
DockerImage

docker_helpers.DockerImage: Docker image object to be used to interact with it.
Source code in src/docker_wrapper/cli.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def __create_image(image_name: str, **kwargs: Dict[str, str]) -> docker_helpers.DockerImage:
    """Instantiate an DockerImage object (or its subclass) for the specified
    Image

    Args:
        image_name (str): Name of the image to create an image for.
        **kwargs (Dict[str, str]): Additional arguments to pass to the DockerImage object.

    Raises:
        typer.Exit: Failure if the image is not found

    Returns:
        docker_helpers.DockerImage: Docker image object to be used to interact with it.
    """
    if image_name not in __WRAPPER_EXTENSIONS:
        typer.echo(f"Unknown Image {image_name}")
        raise typer.Exit(1)
    image = __WRAPPER_EXTENSIONS[image_name](**kwargs)  # type: ignore
    return image

__get_image_name_value(image_name)

Get the string value of image_name

Parameters:

Name Type Description Default
image_name Union[str, EnumClassType]

Value returned by the CLI

 required

Returns:

Name Type Description
str str

Actual value
Source code in src/docker_wrapper/cli.py 
133
134
135
136
137
138
139
140
141
142
143
144
def __get_image_name_value(image_name: Union[str, EnumClassType]) -> str:
    """Get the string value of image_name

    Args:
        image_name (Union[str, EnumClassType]): Value returned by the CLI

    Returns:
        str: Actual value
    """
    if isinstance(image_name, str):
        return image_name
    return str(image_name.value)

create_cli(image_dir=None, env_config_arg=None)

Create the command line interface of the Docker Wrapper

Parameters:

Name Type Description Default
image_dir Optional[str]

Directory where the Docker image Dockerfiles and related code are located. Defaults to None.

 None
env_config_arg Optional[Dict[str, str]]

Configuration environment of the repository. Dictionary containing environment configuration values.

 None

Returns:

Type Description
Typer

typer.Typer: Typer class wit the registered command line arguments. See the main()
Typer

function how an object can be instantiated.
Source code in src/docker_wrapper/cli.py 
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
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def create_cli(
    image_dir: Optional[str] = None, env_config_arg: Optional[Dict[str, str]] = None  #
) -> typer.Typer:
    """Create the command line interface of the Docker Wrapper

    Args:
        image_dir (Optional[str], optional): Directory where the Docker image Dockerfiles and
            related code are located. Defaults to None.
        env_config_arg (Optional[Dict[str, str]], optional): Configuration environment of
            the repository. Dictionary containing environment configuration values.

    Returns:
        typer.Typer: Typer class wit the registered command line arguments. See the main()
        function how an object can be instantiated.
    """
    app = typer.Typer()

    # Initialize the environment configuration based on passed argument
    # Yet expect that the environment configuration could be ovewritten
    # past the creation state. For this reason use the get_env_config() function
    # to get the configuration at execution of each subcommand.
    env_config = env_config_arg or {}
    set_env_config(env_config)

    images = {}
    if image_dir:
        global __WRAPPER_EXTENSIONS
        __WRAPPER_EXTENSIONS = find_extensions(Path(image_dir))
        images = {i: i for i in __WRAPPER_EXTENSIONS.keys()}
        global __DOCKER_IMAGE_CLASS_NAME
        image_names = Enum(__DOCKER_IMAGE_CLASS_NAME, images)  # type: ignore
    else:
        image_names = str  # type: ignore

    @app.callback()
    def main(
        image_dir: Path = typer.Option(image_dir, help="Path where the docker images are located"),
        log_level: LoggingLevel = typer.Option(LoggingLevel.INFO, help="Set logging level"),
    ) -> None:
        """Main command arguments

        Args:
            image_dir (Path, optional): Path where images are stored.
                Defaults to typer.Argument(image_dir).
            log_level (LoggingLevel, optional): Executing logging level. Defaults to
                typer.Option(LoggingLevel.INFO, help="Set logging level").
        """
        # Configure the logging level of the run
        if log_level not in _LOG_LEVEL_STRINGS:
            message = "invalid choice: {0} (choose from {1})".format(log_level, _LOG_LEVEL_STRINGS)
            typer.echo(message)
            typer.Exit(code=1)
        log_level_int = getattr(logging, log_level, logging.INFO)
        # check the logging log_level_choices have not changed from our expected values
        assert isinstance(log_level_int, int)
        logging.basicConfig(level=log_level_int)
        global __WRAPPER_EXTENSIONS
        __WRAPPER_EXTENSIONS = find_extensions(image_dir)

        docker_login_cmd = get_env_config().get("docker_login_command", "")

        if docker_login_cmd:
            try:
                # Log the command safely
                if (
                    "password" not in docker_login_cmd.lower()
                    and "token" not in docker_login_cmd.lower()
                ):
                    logging.info(f"Executing docker login command: {docker_login_cmd}")
                else:
                    logging.info("Executing docker login command (redacted for safety)")

                # Capture and log the output of the command
                result = subprocess.run(
                    docker_login_cmd, shell=True, check=True, text=True, capture_output=True
                )
                logging.info(f"Docker login output: {result.stdout}")
            except subprocess.CalledProcessError as e:
                logging.warning(f"Docker login failed with error: {e}")
                if e.stderr:
                    logging.warning(f"Error details: {e.stderr}")
                logging.warning("Continuing with starting the container")

    @app.command()
    def build(
        image_name: image_names,
    ) -> None:
        """Build a docker image

        Args:
            image_name (image_names): Name of the image to build
        """
        env_config = get_env_config()
        image = __create_image(__get_image_name_value(image_name), **env_config)  # type: ignore
        image.build_image()

    @app.command()
    def push(
        image_name: image_names,
    ) -> None:
        """Push a Docker image to a Docker registry

        Args:
            image_name (image_names): Name of the image to push
        """
        env_config = get_env_config()
        image = __create_image(__get_image_name_value(image_name), **env_config)  # type: ignore
        image.push()

    @app.command()
    def pull(
        image_name: image_names,
    ) -> None:
        """Pull a docker image form a Docker registry

        Args:
            image_name (image_names): Name of the image to pull
        """
        env_config = get_env_config()
        image = __create_image(__get_image_name_value(image_name), **env_config)  # type: ignore
        image.pull()

    @app.command()
    def image_url(
        image_name: image_names,
    ) -> None:
        """URL of the image to use

        Args:
            image_name (image_names): Name of the image to pull
        """
        env_config = get_env_config()
        image = __create_image(__get_image_name_value(image_name), **env_config)  # type: ignore
        print(image.image_url)

    def _start_docker_helper(
        prompt: bool,
        image_name: image_names,
        cmds: List[str],
        mount_home: bool = typer.Option(False, help="Mount the home directory"),
        project_dir: Path = typer.Option(".", help="Path of the repo top-level"),
        network: Optional[str] = typer.Option(None, help="Pass the network information."),
        privileged: bool = typer.Option(False, help="Enable Docker privileged mode"),
        ports: Optional[List[str]] = typer.Option(None, help="Port to forward from Docker"),
        volume: Optional[List[str]] = typer.Option(None, help="Volume to mount"),
        env: Optional[List[str]] = typer.Option(None, help="Environment variables to pass"),
        sudo: bool = typer.Option(True, help="Enable sudo inside the container"),
        mount_host_passwd: bool = typer.Option(
            True,
            help="Mount the host passwd related files inside the container and use the "
            + "`-u` docker option",
        ),
    ) -> None:
        """
        Helper function that start a container and drop the user inside a prompt or run a command
        """
        env_config = get_env_config()
        image = __create_image(__get_image_name_value(image_name), **env_config)  # type: ignore
        image.run(
            prompt=prompt,
            cmds=cmds,
            mount_home=mount_home,
            project_dir=project_dir,
            network=network,
            privileged=privileged,
            ports=ports,
            volumes=volume,
            envs=env,
            enable_sudo=sudo,
            mount_host_passwd=mount_host_passwd,
        )

    @typing.no_type_check
    @app.command()
    @forge.compose(  #
        forge.copy(_start_docker_helper),  #
        forge.delete("prompt"),  #
        forge.delete("cmds"),  #
    )
    def prompt(*args, **kwargs) -> None:
        """Start a docker container and drop the user inside a prompt"""
        _start_docker_helper(prompt=True, cmds=[], *args, **kwargs)  # noqa: B026

    @typing.no_type_check
    @app.command()
    @forge.compose(  #
        forge.copy(_start_docker_helper),  #
        forge.delete("prompt"),  #
    )
    def run(*args, **kwargs):
        """Run the following command inside the container"""
        _start_docker_helper(False, *args, **kwargs)

    for name in images.keys():

        @typing.no_type_check
        @app.command(name)
        @forge.compose(  #
            forge.copy(_start_docker_helper),  #
            forge.delete("prompt"),  #
            forge.delete("image_name"),
        )
        def run_image(
            iname: str = name,
            *args,
            **kwargs,
        ) -> None:
            _start_docker_helper(False, image_name=iname, *args, **kwargs)  # noqa: B026

    return app

find_extensions(image_dir)

Find all available Docker images and extension modules that enable working with them

Parameters:

Name Type Description Default
image_dir Path

Path containing the docker images.

 required

Returns:

Type Description
Dict[str, Type[DockerImage]]

Dict[str, Callable[[], docker_helpers.DockerImage]]: Dictionary of docker image names and
Dict[str, Type[DockerImage]]

object constructors we can call.
Source code in src/docker_wrapper/cli.py 
 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
def find_extensions(image_dir: Path) -> Dict[str, Type[docker_helpers.DockerImage]]:
    """Find all available Docker images and extension modules that enable working with them

    Args:
        image_dir (Path): Path containing the docker images.

    Returns:
        Dict[str, Callable[[], docker_helpers.DockerImage]]: Dictionary of docker image names and
        object constructors we can call.
    """
    # Find all available docker images under the image_dir folder
    # Assume that each folder is a separate docker image and there is no nestting
    # Expected that each folder should have a docker_wrapper_extensions.py Python script
    # and a Docker folder with ant necessary files
    if not image_dir:
        return {}
    docker_images = {}
    for root, dirs, _ in os.walk(os.path.realpath(image_dir), topdown=True):
        for dir in dirs:
            if "Docker" == dir:
                image_name = os.path.basename(root)
                docker_images[image_name] = os.path.realpath(root)
    logging.debug(f"Found docker images: {docker_images}")
    # Load the docker_wrapper_extensions.py, import it as a module and get a callable
    # object for the DockerImage class object that each file contains
    extensions = {}
    for _, ipath in docker_images.items():
        spec = importlib.util.spec_from_file_location(
            "docker_wrapper_extensions", os.path.join(ipath, "docker_wrapper_extensions.py")
        )
        mod = importlib.util.module_from_spec(spec)  # type: ignore
        spec.loader.exec_module(mod)  # type: ignore
        docker_image_classes = []
        for _, obj in inspect.getmembers(mod):
            if inspect.isclass(obj) and issubclass(obj, docker_helpers.DockerImage):
                docker_image_classes.append(obj)
                # docker_image_classes contains the classes that are of type DockerImage
                extensions[obj.NAME] = obj

    return extensions

get_env_config()

Get the configuration of the repository

Returns:

Type Description
Dict[str, str]

Dict[str, str]: Configuration of the repository
Source code in src/docker_wrapper/cli.py 
61
62
63
64
65
66
67
def get_env_config() -> Dict[str, str]:
    """Get the configuration of the repository

    Returns:
        Dict[str, str]: Configuration of the repository
    """
    return LOCAL_ENV_CONFIG

main()

Helper main function

Source code in src/docker_wrapper/cli.py 
359
360
361
362
def main() -> None:
    """Helper main function"""
    app = create_cli()
    app()

set_env_config(config)

Set the configuration of the repository

Parameters:

Name Type Description Default
config Dict[str, str]

Configuration of the repository

 required
Source code in src/docker_wrapper/cli.py 
51
52
53
54
55
56
57
58
def set_env_config(config: Dict[str, str]) -> None:
    """Set the configuration of the repository

    Args:
        config (Dict[str, str]): Configuration of the repository
    """
    global LOCAL_ENV_CONFIG
    LOCAL_ENV_CONFIG = config