Bases: BaseService
A service class for Docker operations.
This class provides methods for managing Docker containers, images,
and building custom Docker environments.
The constructor for the DockerService class.
Parameters:
-
client
(DockerClient | None, default:
None
)
–
The Docker client for container operations.
-
config
(dict[str, str] | None, default:
None
)
–
The Docker configuration dictionary.
Source code in dev_tool/services/docker/service.py
| def __init__(
self,
client: DockerClient | None = None,
config: dict[str, str] | None = None
) -> None:
"""
The constructor for the DockerService class.
:param client: The Docker client for container operations.
:param config: The Docker configuration dictionary.
"""
super().__init__()
self.client = client
self.config = config
|
A method that builds a custom Docker image from a Dockerfile.
Parameters:
-
name
(str)
–
The name for the custom image.
-
dockerfile
(Path)
–
The path to the Dockerfile.
Returns:
-
str
–
The name of the built image.
Raises:
Source code in dev_tool/services/docker/service.py
| def build_custom_image(self, name: str, dockerfile: Path) -> str:
"""
A method that builds a custom Docker image from a Dockerfile.
:param name: The name for the custom image.
:param dockerfile: The path to the Dockerfile.
:return: The name of the built image.
:raises DockerImageError: If image building fails.
"""
assert self.client is not None
try:
context = Path(tempfile.mkdtemp())
try:
self._copy_template(context)
with open(dockerfile, 'r', encoding='utf-8') as handle:
content = handle.read()
variables = self.get_build_template_variable()
template = Template(content)
rendered = template.safe_substitute(variables)
dockerfile_path = context / DockerFileDefault.DB
with open(dockerfile_path, 'w', encoding='utf-8') as handle:
handle.write(rendered)
self.client.images.build(
path=str(context),
dockerfile=str(dockerfile_path),
tag=name,
rm=True,
forcerm=True
)
message = f'Custom Docker image "{name}" built successfully.'
self.notification.normal_text(message)
log.debug(message)
finally:
shutil.rmtree(context, ignore_errors=True)
except Exception as exception:
message = f'Failed to build custom Docker image "{name}"'
log.exception(message)
raise DockerImageError(message) from exception
else:
return name
|
A method that checks for and shuts down containers using a port.
Parameters:
-
port
(int)
–
The port number to check for conflicts.
Raises:
-
DockerContainerError
–
If container shutdown fails.
Source code in dev_tool/services/docker/service.py
| def check_and_shutdown_conflicting_port(self, port: int) -> None:
"""
A method that checks for and shuts down containers using a port.
:param port: The port number to check for conflicts.
:raises DockerContainerError: If container shutdown fails.
"""
assert self.client is not None
try:
containers = self.client.containers.list(all=True)
for container in containers:
attrs = container.attrs or {}
ports = attrs.get('NetworkSettings', {}).get('Ports', {}) or {}
for host_ports in ports.values():
if host_ports is None:
continue
for host_port in host_ports:
if host_port.get('HostPort') == str(port):
message = f'Stopping container "{container.name}" using port {port}'
self.notification.normal_text(message)
log.debug(message)
container.stop()
break
except Exception as exception:
message = f'Failed to check and shutdown containers on port {port}'
log.exception(message)
raise DockerContainerError(message) from exception
|
A method that creates a new Docker container.
Parameters:
-
name
(str)
–
The name for the new container.
-
image
(str)
–
-
host
(int)
–
The host port to bind to.
-
container
(int)
–
The container port to expose.
-
size
(str)
–
The shared memory size for the container.
Raises:
-
DockerContainerError
–
If container creation fails.
Source code in dev_tool/services/docker/service.py
| def create_container(
self,
name: str,
image: str,
host: int,
container: int,
size: str
) -> None:
"""
A method that creates a new Docker container.
:param name: The name for the new container.
:param image: The Docker image to use.
:param host: The host port to bind to.
:param container: The container port to expose.
:param size: The shared memory size for the container.
:raises DockerContainerError: If container creation fails.
"""
assert self.client is not None
assert self.config is not None
try:
self.check_and_shutdown_conflicting_port(host)
try:
self.client.volumes.get(name)
except docker.errors.NotFound:
self.client.volumes.create(name=name)
volumes = {
name: {
'bind': ContainerPathDefault.POSTGRES_DATA,
'mode': 'rw'
}
}
raw_environment = self.config.get('environment', {})
runtime_environment = self.get_runtime_environment_variable()
environment: dict[str, str] = raw_environment if isinstance(raw_environment, dict) else {}
environment.update(runtime_environment)
restart_policy = cast('Any', {'Name': DockerRestartPolicy.UNLESS_STOPPED})
self.client.containers.run(
image,
name=name,
detach=True,
restart_policy=restart_policy,
shm_size=size,
ports={f'{container}/{DockerProtocol.TCP}': host},
volumes=volumes,
environment=environment
)
message = f'Container "{name}" created successfully.'
self.notification.normal_text(message)
except Exception as exception:
message = f'Failed to create container: {name}'
log.exception(message)
raise DockerContainerError(message) from exception
|
A method that ensures a container exists and is running.
Parameters:
-
recreate
(bool, default:
False
)
–
Whether to recreate the container if it exists.
Raises:
-
DockerDesktopNotRunningError
–
If container management fails.
Source code in dev_tool/services/docker/service.py
| def ensure_container(self, recreate: bool = False) -> None:
"""
A method that ensures a container exists and is running.
:param recreate: Whether to recreate the container if it exists.
:raises DockerDesktopNotRunningError: If container management fails.
"""
ExecutionStrategyProvider.reset()
if ExecutionStrategyProvider.is_containerized():
ExecutionStrategyProvider.get().ensure_database(recreate=recreate)
return
self._stop_compose_services()
self.ensure_local_container(recreate=recreate)
|
A method that ensures a local database container exists and is running.
Parameters:
-
recreate
(bool, default:
False
)
–
Whether to recreate the container if it exists.
Raises:
-
DockerDesktopNotRunningError
–
If container management fails.
Source code in dev_tool/services/docker/service.py
| def ensure_local_container(self, recreate: bool = False) -> None:
"""
A method that ensures a local database container exists and is running.
:param recreate: Whether to recreate the container if it exists.
:raises DockerDesktopNotRunningError: If container management fails.
"""
assert self.config is not None
ExecutionStrategyProvider.reset()
self._stop_foreign_container()
self._stop_compose_container()
name = self.config['container_name']
image = self.config['container']
host = int(self.config['host_port'])
container = int(self.config['container_port'])
size = self.config['shm_size']
try:
if recreate:
self.remove_container(name)
docker_config = self.configuration.docker
if docker_config.should_build_custom_image():
dockerfile = docker_config.get_dockerfile_path()
if not self.image_exists(DockerImageDefault.CUSTOM) or recreate:
assert dockerfile is not None
image = self.build_custom_image(DockerImageDefault.CUSTOM, dockerfile)
else:
image = DockerImageDefault.CUSTOM
if self.is_container(name):
current = self.get_container_image(name)
target = self.get_image_id(image)
if current != target:
self.remove_container(name)
if not self.is_container(name):
self.create_container(name, image, host, container, size)
if not self.is_container_running(name):
self.check_and_shutdown_conflicting_port(host)
self.start_container(name)
except DockerDesktopNotRunningError:
raise
except Exception as exception:
message = f'Failed to ensure container: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from exception
|
A method that gets template variables for Docker image building.
Returns:
-
dict[str, str]
–
Dictionary of template variables.
Source code in dev_tool/services/docker/service.py
| def get_build_template_variable(self) -> dict[str, str]:
"""
A method that gets template variables for Docker image building.
:return: Dictionary of template variables.
"""
username, password = self.configuration.get_docker_authorization()
return {
'POSTGRES_USER': username,
'POSTGRES_PASSWORD': password,
}
|
A method that gets the image ID of a container.
Parameters:
-
name
(str)
–
The name of the container.
Returns:
-
str | None
–
The image ID string, or None if not found.
Raises:
-
DockerDesktopNotRunningError
–
If container check fails.
Source code in dev_tool/services/docker/service.py
| def get_container_image(self, name: str) -> str | None:
"""
A method that gets the image ID of a container.
:param name: The name of the container.
:return: The image ID string, or None if not found.
:raises DockerDesktopNotRunningError: If container check fails.
"""
assert self.client is not None
try:
container = self.client.containers.get(name)
except docker.errors.NotFound:
return None
except Exception as exception:
message = f'Failed to get container image: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from exception
else:
image = container.image
return image.id if image else None
|
A method that gets the ID of a Docker image.
Parameters:
Returns:
-
str | None
–
The image ID string, or None if not found.
Raises:
-
DockerDesktopNotRunningError
–
Source code in dev_tool/services/docker/service.py
| def get_image_id(self, name: str) -> str | None:
"""
A method that gets the ID of a Docker image.
:param name: The name of the image.
:return: The image ID string, or None if not found.
:raises DockerDesktopNotRunningError: If image check fails.
"""
assert self.client is not None
try:
image = self.client.images.get(name)
except docker.errors.ImageNotFound:
return None
except Exception as exception:
message = f'Failed to get image: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from exception
else:
return image.id
|
A method that gets runtime environment variables for containers.
Returns:
-
dict[str, str]
–
Dictionary of environment variables.
Source code in dev_tool/services/docker/service.py
| def get_runtime_environment_variable(self) -> dict[str, str]:
"""
A method that gets runtime environment variables for containers.
:return: Dictionary of environment variables.
"""
return {
key: value
for key, value in os.environ.items()
if key.startswith('DATABASE_')
}
|
A method that checks if a Docker image exists.
Parameters:
-
name
(str)
–
The name of the image to check.
Returns:
-
bool
–
True if the image exists, False otherwise.
Source code in dev_tool/services/docker/service.py
| def image_exists(self, name: str) -> bool:
"""
A method that checks if a Docker image exists.
:param name: The name of the image to check.
:return: True if the image exists, False otherwise.
"""
assert self.client is not None
try:
self.client.images.get(name)
except docker.errors.ImageNotFound:
return False
else:
return True
|
A method that checks if a container exists.
Parameters:
-
name
(str)
–
The name of the container to check.
Returns:
-
bool
–
True if the container exists, False otherwise.
Raises:
-
DockerDesktopNotRunningError
–
If container check fails.
Source code in dev_tool/services/docker/service.py
| def is_container(self, name: str) -> bool:
"""
A method that checks if a container exists.
:param name: The name of the container to check.
:return: True if the container exists, False otherwise.
:raises DockerDesktopNotRunningError: If container check fails.
"""
assert self.client is not None
try:
self.client.containers.get(name)
except docker.errors.NotFound:
return False
except Exception as exception:
message = f'Failed to check container: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from exception
else:
return True
|
A method that checks if a container is currently running.
Parameters:
-
name
(str)
–
The name of the container to check.
Returns:
-
bool
–
True if the container is running, False otherwise.
Raises:
-
DockerDesktopNotRunningError
–
If container check fails.
Source code in dev_tool/services/docker/service.py
| def is_container_running(self, name: str) -> bool:
"""
A method that checks if a container is currently running.
:param name: The name of the container to check.
:return: True if the container is running, False otherwise.
:raises DockerDesktopNotRunningError: If container check fails.
"""
assert self.client is not None
try:
container = self.client.containers.get(name)
except docker.errors.NotFound:
return False
except Exception as exception:
message = f'Failed to check container status: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from exception
else:
return container.status == DockerContainerStatus.RUNNING
|
A method that removes a Docker container.
Parameters:
-
name
(str)
–
The name of the container to remove.
Raises:
-
DockerDesktopNotRunningError
–
If container removal fails.
Source code in dev_tool/services/docker/service.py
| def remove_container(self, name: str) -> None:
"""
A method that removes a Docker container.
:param name: The name of the container to remove.
:raises DockerDesktopNotRunningError: If container removal fails.
"""
assert self.client is not None
try:
try:
container = self.client.containers.get(name)
container.stop()
container.remove(v=True)
message = f'Container "{name}" removed successfully.'
self.notification.normal_text(message)
log.debug(message)
except docker.errors.NotFound:
pass
except Exception as exception:
message = f'Failed to remove container: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from exception
|
A method that starts a stopped container.
Parameters:
-
name
(str)
–
The name of the container to start.
Raises:
-
DockerDesktopNotRunningError
–
If container start fails.
Source code in dev_tool/services/docker/service.py
| def start_container(self, name: str) -> None:
"""
A method that starts a stopped container.
:param name: The name of the container to start.
:raises DockerDesktopNotRunningError: If container start fails.
"""
assert self.client is not None
assert self.config is not None
if not self.is_container(name):
return
try:
container = self.client.containers.get(name)
if container.status == DockerContainerStatus.RUNNING:
return
if container.status in [DockerContainerStatus.EXITED, DockerContainerStatus.CREATED]:
self.check_and_shutdown_conflicting_port(int(self.config['host_port']))
container.start()
message = f'Container "{name}" started successfully.'
self.notification.normal_text(message)
except Exception as exception:
message = f'Failed to start container: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from exception
|