containers

Containers

Containers(image: Image, client: Optional[DockerClient] = None, runtime: type(Runtime) = SimpleRuntime)

Manages the creation and lifecycle of Docker container instances derived from a given Docker image.

This class provides methods to create new container instances using a default or customized runtime configuration, run those containers using a specified runtime (default is SimpleRuntime), and then stop and remove them while maintaining a history of container records and their configurations.

Parameters:

• image (Image) –

  An instance of the Image class representing the Docker image to be used.

• client (Optional[DockerClient], default: None ) –

  A Docker client instance. Defaults to docker.from_env().

• runtime (type(Runtime), default: SimpleRuntime ) –

  The runtime class used to manage container operations. Must be a subclass of Runtime. Defaults to SimpleRuntime.

Attributes:

• _client (DockerClient) –

  The Docker client instance.

• _image (Image) –

  The Docker image used for creating containers.

• _runtime_history (Dict[str, Dict[str, Union[Container, str]]]) –

  A dictionary storing records of created container instances and their associated runtime configurations, keyed by a unique container name.

• _runtime (type(Runtime) –

  The runtime class used for managing container lifecycle operations.

Methods:

• get_container –

  bool = False) -> List[Tuple[str, Dict[str, Union[Container, str]]]]: Retrieve container records from the history. If new is True, returns only records whose container status is "created".

• _default_config –

  Generates a default RuntimeConfig based on the image and a generated unique container name.

• _construct_config –

  Constructs a RuntimeConfig by updating the default configuration with provided keyword arguments.

• _add_record –

  Container, config: RuntimeConfig) -> None: Adds a record for a newly created container and its configuration to the runtime history.

• create –

  Creates a new container using the provided configuration keyword arguments, records it in the history, and returns the container.

• run –

  Runs all newly created containers using the specified runtime and returns their records.

• stop –

  Optional[Union[str, Path]] = None) -> List[Tuple[str, Dict[str, Union[Container, str]]]]: Stops all managed containers, collects their logs (saving them to the specified directory), removes them, and returns the updated container records.

Examples:

>>> from ures.docker.image import Image
>>> from ures.docker.containers import Containers
>>> from ures.docker.conf import RuntimeConfig
>>> img = Image("myapp", tag="latest")
>>> containers_manager = Containers(image=img)
>>> container = containers_manager.create(name="myapp-instance")
>>> containers_manager.run()
>>> containers_manager.stop("/tmp/container_logs")

Source code in ures/docker/containers.py

def __init__(
    self,
    image: Image,
    client: Optional[docker.DockerClient] = None,
    runtime: type(Runtime) = SimpleRuntime,
):
    # Ensure that the image is an instance of the Image class
    assert isinstance(image, Image)
    # Ensure that the provided runtime is a subclass of Runtime
    assert isinstance(runtime, type(Runtime))
    self._client = client or docker.from_env()
    self._image = image
    self._runtime_history: Dict[str, Dict[str, Union[Container, str]]] = {}
    self._runtime: type(Runtime) = runtime

image property

image: str

Returns the full image name including the tag.

Returns:

• str ( str ) –

  The full image name.

Examples:

>>> containers_manager.image
'myapp:latest'

name property

name: str

Generates a unique container name.

Returns:

• str –

  A name of the form {image.name}-instance-{unique_id}.

Examples:

>>> containers_manager.name
'myapp-instance-abc123def4'

history property

history: Dict[str, Dict[str, Union[Container, str]]]

Retrieves the history of container records.

Returns:

• Dict[str, Dict[str, Union[Container, str]]] –

  Dict[str, Dict[str, Union[Container, str]]]: A dictionary of container records keyed by unique names.

Examples:

>>> containers_manager.history
{'myapp-instance-abc123def4': {'container': <Container object>, 'config': <RuntimeConfig>}}

get_container

get_container(new: bool = False) -> List[Tuple[str, Dict[str, Union[Container, str]]]]

Retrieve container records from the history.

Parameters:

• new (bool, default: False ) –

  If True, only return container records whose container status is "created". Defaults to False.

Returns:

• List[Tuple[str, Dict[str, Union[Container, str]]]] –

  List[Tuple[str, Dict[str, Union[Container, str]]]]: A list of tuples containing the container's unique name

• List[Tuple[str, Dict[str, Union[Container, str]]]] –

  and its record.

Examples:

>>> records = containers_manager.get_container(new=True)

Source code in ures/docker/containers.py

def get_container(
    self, new: bool = False
) -> List[Tuple[str, Dict[str, Union[Container, str]]]]:
    """
    Retrieve container records from the history.

    Args:
        new (bool, optional): If True, only return container records whose container status is "created".
            Defaults to False.

    Returns:
        List[Tuple[str, Dict[str, Union[Container, str]]]]: A list of tuples containing the container's unique name
        and its record.

    Examples:
        >>> records = containers_manager.get_container(new=True)
    """
    if new:
        containers = list(
            filter(
                lambda x: x[1]["container"].status == "created",
                self._runtime_history.items(),
            )
        )
    else:
        containers = list(self._runtime_history.items())
    return containers

create

create(**kwargs) -> Container

Create a new container instance using the specified configuration.

Parameters:

• **kwargs (Any, default: {} ) –

  RuntimeConfig fields to override, such as name or detach.

Returns:

• Container ( Container ) –

  The newly created container instance.

Examples:

>>> container = containers_manager.create(name="instance1")

Source code in ures/docker/containers.py

def create(self, **kwargs) -> Container:
    """
    Create a new container instance using the specified configuration.

    Args:
        **kwargs (Any): RuntimeConfig fields to override, such as name or detach.

    Returns:
        Container: The newly created container instance.

    Examples:
        >>> container = containers_manager.create(name="instance1")
    """
    _container = Container(image=self._image, client=self._client)
    _conf = self._construct_config(**kwargs)
    _container.create(config=_conf, tag=None)
    self._add_record(container=_container, config=_conf)
    return _container

run

run() -> List[Tuple[str, Dict[str, Union[Container, str]]]]

Run all newly created containers using the specified runtime.

Returns:

• List[Tuple[str, Dict[str, Union[Container, str]]]] –

  List[Tuple[str, Dict[str, Union[Container, str]]]]: The list of container records that were run.

Examples:

>>> records = containers_manager.run()

Source code in ures/docker/containers.py

def run(self) -> List[Tuple[str, Dict[str, Union[Container, str]]]]:
    """
    Run all newly created containers using the specified runtime.

    Returns:
        List[Tuple[str, Dict[str, Union[Container, str]]]]: The list of container records that were run.

    Examples:
        >>> records = containers_manager.run()
    """
    new_containers = self.get_container(new=True)
    containers = [_c[1]["container"] for _c in new_containers]
    _runtime = self._runtime(containers=containers)
    _runtime.run()
    return new_containers

stop

stop(log_dir: Optional[Union[str, Path]] = None) -> List[Tuple[str, Dict[str, Union[Container, str]]]]

Stop all managed containers, collect their logs (saving them to the specified directory), and remove them.

Parameters:

• log_dir (Optional[Union[str, Path]], default: None ) –

  The directory where container logs will be saved. If not provided, a default temporary directory is used.

Returns:

• List[Tuple[str, Dict[str, Union[Container, str]]]] –

  List[Tuple[str, Dict[str, Union[Container, str]]]]: The updated container records after stopping and removal.

Examples:

>>> records = containers_manager.stop("/tmp/container_logs")

Source code in ures/docker/containers.py

def stop(
    self, log_dir: Optional[Union[str, Path]] = None
) -> List[Tuple[str, Dict[str, Union[Container, str]]]]:
    """
    Stop all managed containers, collect their logs (saving them to the specified directory), and remove them.

    Args:
        log_dir (Optional[Union[str, Path]], optional): The directory where container logs will be saved.
            If not provided, a default temporary directory is used.

    Returns:
        List[Tuple[str, Dict[str, Union[Container, str]]]]: The updated container records after stopping and removal.

    Examples:
        >>> records = containers_manager.stop("/tmp/container_logs")
    """
    log_dir = log_dir or get_temp_dir_with_specific_path("container-logs")
    log_dir = Path(log_dir)
    containers = [_c[1]["container"] for _c in self.get_container()]
    _runtime = self._runtime(containers=containers)
    _runtime.stop()
    _runtime.logs(output_dir=log_dir)
    _runtime.remove()
    return self.get_container()