runtime

Runtime

Runtime(containers: List[Container], interval: float = 0.5)

Bases: ABC

Initializes the Runtime with a list of containers.

Parameters:

• containers (List[Container]) –

  A list of Container instances to manage. Each container must have been created (i.e. container.is_created is True).

Examples:

>>> runtime = SomeRuntime([container1, container2])

Source code in ures/docker/runtime.py

def __init__(self, containers: List[Container], interval: float = 0.5):
    """
    Initializes the Runtime with a list of containers.

    Args:
        containers (List[Container]): A list of Container instances to manage.
            Each container must have been created (i.e. container.is_created is True).

    Examples:
        >>> runtime = SomeRuntime([container1, container2])
    """
    assert all([container.is_created for container in containers]) is True
    self._containers: List[Container] = containers
    self._interval = interval

run abstractmethod

run(*args, **kwargs)

Run all managed containers.

Examples:

>>> runtime.run()

Source code in ures/docker/runtime.py

@abstractmethod
def run(self, *args, **kwargs):
    """
    Run all managed containers.

    Examples:
        >>> runtime.run()
    """
    pass

stop abstractmethod

stop(*args, **kwargs)

Stop all managed containers.

Examples:

>>> runtime.stop()

Source code in ures/docker/runtime.py

@abstractmethod
def stop(self, *args, **kwargs):
    """
    Stop all managed containers.

    Examples:
        >>> runtime.stop()
    """
    pass

remove abstractmethod

remove(*args, **kwargs)

Remove all managed containers.

Examples:

>>> runtime.remove()

Source code in ures/docker/runtime.py

@abstractmethod
def remove(self, *args, **kwargs):
    """
    Remove all managed containers.

    Examples:
        >>> runtime.remove()
    """
    pass

logs abstractmethod

logs(output_dir: Union[str, Path], *args, **kwargs)

Retrieve logs from all managed containers and save them to the specified directory.

Parameters:

• output_dir (Union[str, Path]) –

  The directory where log files should be saved.

Examples:

>>> runtime.logs("/tmp/container_logs")

Source code in ures/docker/runtime.py

@abstractmethod
def logs(self, output_dir: Union[str, Path], *args, **kwargs):
    """
    Retrieve logs from all managed containers and save them to the specified directory.

    Args:
        output_dir (Union[str, Path]): The directory where log files should be saved.

    Examples:
        >>> runtime.logs("/tmp/container_logs")
    """
    pass

SimpleRuntime

SimpleRuntime(containers: List[Container], interval: float = 0.5)

Bases: Runtime

Source code in ures/docker/runtime.py

def __init__(self, containers: List[Container], interval: float = 0.5):
    """
    Initializes the Runtime with a list of containers.

    Args:
        containers (List[Container]): A list of Container instances to manage.
            Each container must have been created (i.e. container.is_created is True).

    Examples:
        >>> runtime = SomeRuntime([container1, container2])
    """
    assert all([container.is_created for container in containers]) is True
    self._containers: List[Container] = containers
    self._interval = interval

run

run(*args, **kwargs)

Runs each container by calling its run() method. If the container becomes running, wait() is called; otherwise, an error is logged.

Examples:

>>> runtime = SimpleRuntime([container1, container2])
>>> runtime.run()

Source code in ures/docker/runtime.py

def run(self, *args, **kwargs):
    """
    Runs each container by calling its run() method. If the container becomes running,
    wait() is called; otherwise, an error is logged.

    Examples:
        >>> runtime = SimpleRuntime([container1, container2])
        >>> runtime.run()
    """
    for container in tqdm.tqdm(self._containers):
        container.run()
        self._regular_delay()
        if container.is_running is False:
            logging.error(f"[Failed] {container.image_name} failed to start")
            continue
        else:
            container.wait()

stop

stop(*args, **kwargs)

Stops each container by calling its stop() method.

Examples:

>>> runtime.stop()

Source code in ures/docker/runtime.py

def stop(self, *args, **kwargs):
    """
    Stops each container by calling its stop() method.

    Examples:
        >>> runtime.stop()
    """
    for container in tqdm.tqdm(self._containers):
        container.stop()
        self._regular_delay()

remove

remove(*args, **kwargs)

Removes each container by calling its remove() method.

Examples:

>>> runtime.remove()

Source code in ures/docker/runtime.py

def remove(self, *args, **kwargs):
    """
    Removes each container by calling its remove() method.

    Examples:
        >>> runtime.remove()
    """
    for container in tqdm.tqdm(self._containers):
        container.remove()
        self._regular_delay()

logs

logs(output_dir: Union[str, Path], *args, **kwargs)

Retrieves logs from each container and writes them to a "logs.txt" file in a directory named after the container's image (with ":" replaced by "-").

Parameters:

• output_dir (Union[str, Path]) –

  The directory where logs should be stored.

Examples:

>>> runtime.logs("/tmp/container_logs")

Source code in ures/docker/runtime.py

def logs(self, output_dir: Union[str, Path], *args, **kwargs):
    """
    Retrieves logs from each container and writes them to a "logs.txt" file in a directory
    named after the container's image (with ":" replaced by "-").

    Args:
        output_dir (Union[str, Path]): The directory where logs should be stored.

    Examples:
        >>> runtime.logs("/tmp/container_logs")
    """
    output_dir = Path(output_dir)
    for container in tqdm.tqdm(self._containers):
        container_dir = output_dir / container.image_name.replace(":", "-")
        container_dir.mkdir(
            parents=True, exist_ok=True
        )  # Ensure the directory exists
        log = container.logs()
        with open(container_dir / "logs.txt", "w") as f:
            f.write(log.decode("utf-8"))