src.asqi.container_manager¶

Attributes¶

`logger`
`INPUT_MOUNT_PATH`
`OUTPUT_MOUNT_PATH`

Exceptions¶

`ManifestExtractionError`	Exception raised when manifest extraction fails.
`MissingImageException`	Exception raised when required Docker images are missing.
`MountExtractionError`	Exception raised when extracting mounts from args fails.

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¶

exception src.asqi.container_manager.ManifestExtractionError(message: str, error_type: str, original_error: Exception | None = None)¶

Bases: Exception

Exception raised when manifest extraction fails.

error_type¶

original_error = None¶

exception src.asqi.container_manager.MissingImageException¶

Bases: Exception

Exception raised when required Docker images are missing.

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:: MissingImageException: 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

exception src.asqi.container_manager.MountExtractionError¶

Bases: Exception

Exception raised when extracting mounts from args fails.

src.asqi.container_manager.run_container_with_args(image: str, args: List[str], container_config: asqi.config.ContainerConfig, environment: Dict[str, str] | None = None) → 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
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.