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 creates and tests a Docker client connection.
Returns:
-
DockerClient | None
–
A Docker client if connection succeeds, None otherwise.
Source code in dev_tool/services/docker/service.py
| @staticmethod
def get_client() -> DockerClient | None:
"""
A method that creates and tests a Docker client connection.
:return: A Docker client if connection succeeds, None otherwise.
"""
try:
client = docker.from_env()
except Exception:
return None
else:
if client and client.ping():
return client
return None
|
A method that builds a custom Docker image from a Dockerfile.
Parameters:
-
name
(str)
–
The name to assign to the built 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 to assign to the built 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
if str(dockerfile).find('templates') != -1:
context = Path(tempfile.mkdtemp())
else:
context = dockerfile.parent
temporary = None
try:
self._copy_templates(context)
variables = self.get_build_template_variables()
with open(dockerfile, 'r', encoding='utf-8') as handle:
content = handle.read()
template = Template(content)
prepared = template.safe_substitute(variables)
temporary = context / 'Dockerfile.temp'
with open(temporary, 'w', encoding='utf-8') as handle:
handle.write(prepared)
message = f'Building custom Docker image: {name}'
self.notification.normal_text(message)
log.debug(message)
self.client.images.build(
path=str(context),
dockerfile='Dockerfile.temp',
tag=name,
rm=True,
forcerm=True,
nocache=False
)
message = f'Successfully built custom image: {name}'
self.notification.normal_text(message)
except Exception:
message = f'Failed to build custom Docker image: {name}'
log.exception(message)
raise DockerImageError(message) from None
else:
return name
finally:
if temporary and temporary.exists():
temporary.unlink()
for filename in ['initialization.sql', 'extensions.sh', 'entrypoint.sh']:
filepath = context / filename
if filepath.exists():
filepath.unlink()
if str(dockerfile).find('templates') != -1:
shutil.rmtree(context, ignore_errors=True)
|
A method that checks for and shuts down containers using a specific port.
Parameters:
-
port
(int)
–
The port number to check for conflicts.
Raises:
-
DockerDesktopNotRunningError
–
If port conflict resolution 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 specific port.
:param port: The port number to check for conflicts.
:raises DockerDesktopNotRunningError: If port conflict resolution fails.
"""
assert self.client is not None
try:
containers = cast(
'list[Container]',
self.client.containers.list()
)
for container in containers:
attrs = cast(
'dict[str, Any]',
container.attrs
)
if attrs and 'NetworkSettings' in attrs:
settings = attrs['NetworkSettings']
if settings and 'Ports' in settings:
ports = settings['Ports']
if ports:
for bindings in ports.values():
if bindings:
for binding in bindings:
if binding['HostPort'] == str(port):
container.stop()
return
except Exception:
message = f'Failed to check and shutdown conflicting port: {port}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
|
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': '/var/lib/postgresql/data',
'mode': 'rw'
}
}
environment = self.config.get('environment', {})
runtime_environment = self.get_runtime_environment_variables()
if isinstance(environment, dict):
environment.update(runtime_environment)
self.client.containers.run(
image,
name=name,
detach=True,
restart_policy={'Name': DockerRestartPolicy.UNLESS_STOPPED},
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:
message = f'Failed to create container: {name}'
log.exception(message)
raise DockerContainerError(message) from None
|
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.
"""
assert self.config is not None
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 = CONTEXT.configuration.docker
if docker_config.should_build_custom_image():
dockerfile = docker_config.get_dockerfile_path()
custom = 'stratus:shared'
if not self.image_exists(custom) or recreate:
assert dockerfile is not None
image = self.build_custom_image(custom, dockerfile)
else:
image = 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.start_container(name)
except DockerDesktopNotRunningError:
raise
except Exception:
message = f'Failed to ensure container: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
|
A method that gets template variables for Docker image building.
Returns:
-
dict[str, str]
–
A dictionary of build-time template variables.
Raises:
-
DockerConfigurationError
–
If variable retrieval fails.
Source code in dev_tool/services/docker/service.py
| def get_build_template_variables(self) -> dict[str, str]:
"""
A method that gets template variables for Docker image building.
:return: A dictionary of build-time template variables.
:raises DockerConfigurationError: If variable retrieval fails.
"""
try:
docker_config = CONTEXT.configuration.get_docker_config()
return {
'POSTGRES_VERSION': str(docker_config.get('postgres-version', '14')),
}
except Exception:
message = 'Failed to get build template variables'
log.exception(message)
raise DockerConfigurationError(message) from None
|
A method that gets the image ID used by a container.
Parameters:
-
name
(str)
–
The name of the container.
Returns:
-
str | None
–
The image ID, or None if container doesn't exist.
Raises:
-
DockerDesktopNotRunningError
–
If image retrieval 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 used by a container.
:param name: The name of the container.
:return: The image ID, or None if container doesn't exist.
:raises DockerDesktopNotRunningError: If image retrieval fails.
"""
assert self.client is not None
if not self.is_container(name):
return None
try:
container = cast(
'Container',
self.client.containers.get(name)
)
except Exception:
message = f'Failed to get container image for: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
else:
image = container.image
if image is not None:
return image.id
return None
|
A method that gets the ID of a Docker image.
Parameters:
-
name
(str)
–
The name or tag of the image.
Returns:
-
str | None
–
The image ID, or None if image doesn't exist.
Raises:
-
DockerDesktopNotRunningError
–
If image ID retrieval fails.
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 or tag of the image.
:return: The image ID, or None if image doesn't exist.
:raises DockerDesktopNotRunningError: If image ID retrieval fails.
"""
assert self.client is not None
try:
image = cast(
'Image',
self.client.images.get(name)
)
except docker.errors.ImageNotFound:
return None
except Exception:
message = f'Failed to get image ID for: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
else:
return image.id
|
A method that gets runtime environment variables for Docker containers.
Returns:
-
dict[str, str]
–
A dictionary of runtime environment variables.
Raises:
-
DockerConfigurationError
–
If variable retrieval fails.
Source code in dev_tool/services/docker/service.py
| def get_runtime_environment_variables(self) -> dict[str, str]:
"""
A method that gets runtime environment variables for Docker containers.
:return: A dictionary of runtime environment variables.
:raises DockerConfigurationError: If variable retrieval fails.
"""
try:
project = CONTEXT.configuration.get_project_name()
return {
'POSTGRES_USER': os.getenv('DATABASE_USER', 'stratus'),
'POSTGRES_PASSWORD': os.getenv('DATABASE_PASSWORD', 'stratus'),
'POSTGRES_DB': os.getenv('DATABASE_NAME', project),
}
except Exception:
message = 'Failed to get runtime environment variables'
log.exception(message)
raise DockerConfigurationError(message) from None
|
A method that gets template variables for Docker configuration.
Returns:
-
dict[str, str]
–
A dictionary of template variables.
Raises:
-
DockerConfigurationError
–
If variable retrieval fails.
Source code in dev_tool/services/docker/service.py
| def get_template_variables(self) -> dict[str, str]:
"""
A method that gets template variables for Docker configuration.
:return: A dictionary of template variables.
:raises DockerConfigurationError: If variable retrieval fails.
"""
try:
project = CONTEXT.configuration.get_project_name()
docker_config = CONTEXT.configuration.get_docker_config()
return {
'PROJECT_NAME': project,
'POSTGRES_VERSION': str(docker_config.get('postgres-version', '14')),
'POSTGRES_USER': os.getenv('DATABASE_USER', 'stratus'),
'POSTGRES_PASSWORD': os.getenv('DATABASE_PASSWORD', 'stratus'),
'POSTGRES_DB': os.getenv('DATABASE_NAME', project),
}
except Exception:
message = 'Failed to get template variables'
log.exception(message)
raise DockerConfigurationError(message) from None
|
A method that checks if a Docker image exists.
Parameters:
-
name
(str)
–
The name or tag of the image.
Returns:
-
bool
–
True if the image exists, False otherwise.
Raises:
-
DockerDesktopNotRunningError
–
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 or tag of the image.
:return: True if the image exists, False otherwise.
:raises DockerDesktopNotRunningError: If image check fails.
"""
assert self.client is not None
try:
self.client.images.get(name)
except docker.errors.ImageNotFound:
return False
except Exception:
message = f'Failed to check if image exists: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
else:
return True
|
A method that checks if a container exists.
Parameters:
-
name
(str)
–
The name of the container.
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.
:return: True if the container exists, False otherwise.
:raises DockerDesktopNotRunningError: If container check fails.
"""
assert self.client is not None
try:
containers = cast(
'list[Container]',
self.client.containers.list(all=True)
)
return any(container.name == name for container in containers)
except Exception:
message = f'Failed to check if container exists: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
|
A method that checks if a container is currently running.
Parameters:
-
name
(str)
–
The name of the container.
Returns:
-
bool
–
True if the container is running, False otherwise.
Raises:
-
DockerDesktopNotRunningError
–
If container status 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.
:return: True if the container is running, False otherwise.
:raises DockerDesktopNotRunningError: If container status check fails.
"""
assert self.client is not None
if not self.is_container(name):
return False
try:
container = cast(
'Container',
self.client.containers.get(name)
)
except Exception:
message = f'Failed to check if container is running: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
else:
return container.status == DockerContainerStatus.RUNNING
|
A method that checks if Docker is running and accessible.
Returns:
-
bool
–
True if Docker is running, False otherwise.
Source code in dev_tool/services/docker/service.py
| def is_docker_running(self) -> bool:
"""
A method that checks if Docker is running and accessible.
:return: True if Docker is running, False otherwise.
"""
assert self.client is not None
try:
self.client.ping()
except docker.errors.DockerException:
message = 'Docker is not running or unreachable.'
log.exception(message)
return False
except Exception:
message = 'Unexpected error checking Docker status'
log.exception(message)
return False
else:
return True
|
A method that prepares a Dockerfile with template variable substitution.
Parameters:
-
dockerfile
(Path)
–
The path to the Dockerfile template.
-
variables
(dict[str, str])
–
The template variables to substitute.
Returns:
-
str
–
The processed Dockerfile content.
Raises:
-
DockerDesktopNotRunningError
–
If Dockerfile preparation fails.
Source code in dev_tool/services/docker/service.py
| def prepare_dockerfile(self, dockerfile: Path, variables: dict[str, str]) -> str:
"""
A method that prepares a Dockerfile with template variable substitution.
:param dockerfile: The path to the Dockerfile template.
:param variables: The template variables to substitute.
:return: The processed Dockerfile content.
:raises DockerDesktopNotRunningError: If Dockerfile preparation fails.
"""
try:
with open(dockerfile, 'r', encoding='utf-8') as handle:
content = handle.read()
template = Template(content)
return template.safe_substitute(variables)
except Exception:
message = f'Failed to prepare Dockerfile: {dockerfile}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
|
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
if not self.is_container(name):
return
try:
container = cast(
'Container',
self.client.containers.get(name)
)
container.remove(force=True)
message = f'Container "{name}" removed successfully.'
self.notification.normal_text(message)
volume = cast(
'Volume',
self.client.volumes.get(name)
)
volume.remove(force=True)
message = f'Volume "{name}" removed successfully.'
self.notification.normal_text(message)
except docker.errors.NotFound:
pass
except Exception:
message = f'Failed to remove container: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
|
A method that starts a Docker 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 Docker 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 = cast(
'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:
message = f'Failed to start container: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
|
A method that stops a Docker container.
Parameters:
-
name
(str)
–
The name of the container to stop.
Raises:
-
DockerDesktopNotRunningError
–
Source code in dev_tool/services/docker/service.py
| def stop_container(self, name: str) -> None:
"""
A method that stops a Docker container.
:param name: The name of the container to stop.
:raises DockerDesktopNotRunningError: If container stop fails.
"""
assert self.client is not None
if not self.is_container(name):
return
try:
container = cast(
'Container',
self.client.containers.get(name)
)
container.stop()
message = f'Container "{name}" stopped successfully.'
self.notification.normal_text(message)
except Exception:
message = f'Failed to stop container: {name}'
log.exception(message)
raise DockerDesktopNotRunningError(message) from None
|