src.asqi.container_manager

Attributes

logger
INPUT_MOUNT_PATH
OUTPUT_MOUNT_PATH
TIMEOUT_EXCEPTIONS

Functions

docker_client()	Context manager for Docker client with proper cleanup.
check_images_availability(→ Dict[str, bool])	Check if Docker images are available locally.
pull_images(images)	Pull Docker images from registry if not available locally.
extract_manifest_from_image(...)	Extract and parse manifest.yaml from a Docker image.
run_container_with_args(→ Dict[str, Any])	Run a Docker container with specified arguments and return results.
shutdown_containers(→ None)	Force-remove any containers that are still tracked as active.

Module Contents

src.asqi.container_manager.logger

src.asqi.container_manager.INPUT_MOUNT_PATH

src.asqi.container_manager.OUTPUT_MOUNT_PATH

src.asqi.container_manager.TIMEOUT_EXCEPTIONS

src.asqi.container_manager.docker_client()

Context manager for Docker client with proper cleanup.

src.asqi.container_manager.check_images_availability(images: List[str]) → Dict[str, bool]

Check if Docker images are available locally. - Tries to fetch each image by exact name:tag.

Returns:

Dict mapping image -> True/False indicating availability

src.asqi.container_manager.pull_images(images: List[str])

Pull Docker images from registry if not available locally.

Args:

images: List of image references (e.g., ‘ubuntu:22.04’, ‘user/repo:tag’)

Returns:

None on success

Raises:

MissingImageError: If images cannot be pulled (includes alternative suggestions) ConnectionError: If unable to connect to Docker daemon

src.asqi.container_manager.extract_manifest_from_image(image: str, manifest_path: str = '/app/manifest.yaml') → asqi.schemas.Manifest | None

Extract and parse manifest.yaml from a Docker image.

Args:

image: Docker image name manifest_path: Path to manifest file inside container

Returns:

Parsed Manifest object or None if extraction fails

Raises:

ManifestExtractionError: If extraction fails with detailed error information

src.asqi.container_manager.run_container_with_args(image: str, args: List[str], container_config: asqi.config.ContainerConfig, environment: Dict[str, str] | None = None, name: str | None = None, workflow_id: str = '') → Dict[str, Any]

Run a Docker container with specified arguments and return results.

Args:

image: Docker image to run args: Command line arguments to pass to container container_config: Container execution configurations environment: Optional dictionary of environment variables to pass to container name: Optional name for the container (will be used as container name in Docker) workflow_id: Workflow identifier to uniquely associate the container with a workflow

Returns:

Dictionary with execution results including exit_code, output, success, etc.

src.asqi.container_manager.shutdown_containers() → None

Force-remove any containers that are still tracked as active.

This is intended to run during atexit or signal handling to ensure worker containers do not linger if the main process is interrupted.