Download Management

download

SoftManager class to manage file downloads with a progress display.

This class downloads files from URLs with progress tracking rendered through the :mod:eos_downloader.helpers reporter layer, which selects a rich (terminal) or plain (CI / non-TTY) display automatically.

Functions:

Name	Description
download_file	Downloads a file from the given URL to the specified path with a progress display.

Attributes:

Name	Type	Description
None

Example

>>> downloader = SoftManager()
>>> result = downloader.download_file(
...     url='http://example.com/file.zip',
...     file_path='/downloads',
...     filename='file.zip',
... )

SoftManager

SoftManager(
    dry_run: bool = False,
    force_download: bool = False,
    console: Optional[Console] = None,
)

SoftManager helps to download files from a remote location.

This class provides methods to download files using either a simple progress bar or a rich interface with enhanced visual feedback.

Examples:

>>> downloader = SoftManager()
>>> downloader.download_file(
...     url="http://example.com/file.txt",
...     file_path="/tmp",
...     filename="file.txt"
... )
'/tmp/file.txt'

Parameters:

Name	Type	Description	Default
dry_run	bool	If True, simulate operations without executing them, by default False	False
force_download	bool	If True, bypass cache and force download/import, by default False	False
console	Console	Shared Rich console used to render the progress display. When omitted, a default Console is created (its terminal detection then drives progress="auto").	None

Source code in eos_downloader/logics/download.py

def __init__(
    self,
    dry_run: bool = False,
    force_download: bool = False,
    console: Optional[Console] = None,
) -> None:
    """
    Initialize SoftManager.

    Parameters
    ----------
    dry_run : bool, optional
        If True, simulate operations without executing them, by default False
    force_download : bool, optional
        If True, bypass cache and force download/import, by default False
    console : Console, optional
        Shared Rich console used to render the progress display. When omitted,
        a default ``Console`` is created (its terminal detection then drives
        ``progress="auto"``).
    """
    self.file: Dict[str, Union[str, None]] = {}
    self.file["name"] = None
    self.file["md5sum"] = None
    self.file["sha512sum"] = None
    self.dry_run = dry_run
    self.force_download = force_download
    self.console = console if console is not None else Console()
    logger.info(
        f"SoftManager initialized{' in dry-run mode' if dry_run else ''}{' with force download' if force_download else ''}"
    )

checksum

checksum(
    check_type: Literal["md5sum", "sha512sum", "md5"],
) -> bool

Verifies the integrity of a downloaded file using a specified checksum algorithm.

Parameters:

Name	Type	Description	Default
check_type	Literal['md5sum', 'sha512sum', 'md5']	The type of checksum to perform. Currently supports ‘md5sum’ or ‘sha512sum’.	required

Returns:

Type	Description
bool	True if the checksum verification passes.

Raises:

Type	Description
ValueError	If the calculated checksum does not match the expected checksum.
FileNotFoundError	If either the checksum file or the target file cannot be found.

Examples:

>>> client.checksum('sha512sum')  # Returns True if checksum matches

Source code in eos_downloader/logics/download.py

def checksum(self, check_type: Literal["md5sum", "sha512sum", "md5"]) -> bool:
    """
    Verifies the integrity of a downloaded file using a specified checksum algorithm.

    Parameters
    ----------
    check_type : Literal['md5sum', 'sha512sum', 'md5']
        The type of checksum to perform. Currently supports 'md5sum' or 'sha512sum'.

    Returns
    -------
    bool
        True if the checksum verification passes.

    Raises
    ------
    ValueError
        If the calculated checksum does not match the expected checksum.
    FileNotFoundError
        If either the checksum file or the target file cannot be found.

    Examples
    --------
    >>> client.checksum('sha512sum')  # Returns True if checksum matches
    """
    logger.info(f"Checking checksum for {self.file['name']} using {check_type}")

    if self.dry_run:
        logger.debug("Dry-run mode enabled, skipping checksum verification")
        return True

    if check_type == "sha512sum":
        hash_sha512 = hashlib.sha512()
        hash512sum = self.file["sha512sum"]
        file_name = self.file["name"]

        logger.debug(f"checksum sha512sum file is: {hash512sum}")

        if file_name is None or hash512sum is None:
            logger.error("File or checksum not found")
            raise ValueError("File or checksum not found")

        with open(hash512sum, "r", encoding="utf-8") as f:
            hash_expected = f.read().split()[0]
        with open(file_name, "rb") as f:
            while True:
                chunk = f.read(4096)
                if not chunk:
                    break
                hash_sha512.update(chunk)
        if hash_sha512.hexdigest() != hash_expected:
            logger.error(
                f"Checksum failed for {self.file['name']}: computed {hash_sha512.hexdigest()} - expected {hash_expected}"
            )
            raise ValueError("Incorrect checksum")
        return True

    if check_type in ["md5sum", "md5"]:
        md5sum_file = self.file["md5sum"]
        file_name = self.file["name"]

        if md5sum_file is None:
            raise ValueError(f"md5sum is not found: {md5sum_file}")

        with open(md5sum_file, "r", encoding="utf-8") as f:
            hash_expected = f.read().split()[0]

        if hash_expected is None:
            raise ValueError("MD5Sum is empty, cannot compute file.")

        if file_name is None:
            raise ValueError("Filename is None. Please fix it")

        if not self._compute_hash_md5sum(file_name, hash_expected=hash_expected):
            logger.error(
                f"Checksum failed for {self.file['name']}: expected {hash_expected}"
            )

            raise ValueError("Incorrect checksum")

        return True

    logger.error(f"Checksum type {check_type} not yet supported")
    raise ValueError(f"Checksum type {check_type} not yet supported")

download_file

download_file(
    url: str,
    file_path: str,
    filename: str,
    *,
    progress: ProgressMode = "auto",
    rich_interface: Optional[bool] = None,
    force: bool = False
) -> Union[None, str]

Downloads a file from a URL with caching support.

Parameters:

Name	Type	Description	Default
url	str	The URL from which to download the file.	required
file_path	str	The directory path where the file should be saved.	required
filename	str	The name to be given to the downloaded file.	required
progress	ProgressMode	Progress rendering mode: "auto" (default, picks rich on a TTY and plain otherwise), "rich", "plain" or "none".	'auto'
rich_interface	bool	Deprecated. Use progress instead. False maps to progress="plain" and True maps to progress="auto".	None
force	bool	If True, download even if file exists locally. Defaults to False.	False

Returns:

Type	Description
Union[None, str]	Path to the downloaded or cached file, or None on error.

Examples:

>>> manager = SoftManager()
>>> manager.download_file(
...     url="https://example.com/file.swi",
...     file_path="/downloads",
...     filename="EOS-4.29.3M.swi"
... )
'/downloads/EOS-4.29.3M.swi'

Source code in eos_downloader/logics/download.py

def download_file(
    self,
    url: str,
    file_path: str,
    filename: str,
    *,
    progress: ProgressMode = "auto",
    rich_interface: Optional[bool] = None,
    force: bool = False,
) -> Union[None, str]:
    """
    Downloads a file from a URL with caching support.

    Parameters
    ----------
    url : str
        The URL from which to download the file.
    file_path : str
        The directory path where the file should be saved.
    filename : str
        The name to be given to the downloaded file.
    progress : ProgressMode, optional
        Progress rendering mode: ``"auto"`` (default, picks rich on a TTY and
        plain otherwise), ``"rich"``, ``"plain"`` or ``"none"``.
    rich_interface : bool, optional
        Deprecated. Use ``progress`` instead. ``False`` maps to
        ``progress="plain"`` and ``True`` maps to ``progress="auto"``.
    force : bool, optional
        If True, download even if file exists locally. Defaults to False.

    Returns
    -------
    Union[None, str]
        Path to the downloaded or cached file, or None on error.

    Examples
    --------
    >>> manager = SoftManager()
    >>> manager.download_file(
    ...     url="https://example.com/file.swi",
    ...     file_path="/downloads",
    ...     filename="EOS-4.29.3M.swi"
    ... )
    '/downloads/EOS-4.29.3M.swi'
    """
    mode = _resolve_progress_mode(progress, rich_interface)
    full_path = Path(file_path) / filename

    # Check cache unless force flag is set
    if not force and not self.force_download:
        if full_path.exists():
            logger.info(f"Using cached file: {full_path}")
            return str(full_path)

    # Log download action
    logger.info(
        f"{'[DRY-RUN] Would download' if self.dry_run else 'Downloading'} {filename} from {url}"
    )

    # Handle dry-run mode
    if self.dry_run:
        return os.path.join(file_path, filename)

    # Proceed with download
    if url is not False:
        reporter = resolve_reporter(
            mode, self.console, title=f"Downloading {filename}"
        )
        download_files_concurrently([(url, str(full_path), filename)], reporter)
        return os.path.join(file_path, filename)

    logger.error(f"Cannot download file {file_path}")
    return None

downloads

downloads(
    object_arista: AristaXmlObjects,
    file_path: str,
    *,
    progress: ProgressMode = "auto",
    rich_interface: Optional[bool] = None
) -> tuple[str, bool]

Downloads files from Arista EOS server with caching support.

Downloads the EOS image and optional md5/sha512 files based on the provided EOS XML object. Each file is downloaded to the specified path with appropriate filenames. Uses cache to skip already downloaded files unless force_download is enabled. The files of a single download are fetched concurrently and rendered in one shared progress display.

Parameters:

Name	Type	Description	Default
object_arista	AristaXmlObjects	Object containing EOS image and hash file URLs.	required
file_path	str	Directory path where files should be downloaded.	required
progress	ProgressMode	Progress rendering mode: "auto" (default), "rich", "plain" or "none".	'auto'
rich_interface	bool	Deprecated. Use progress instead. False maps to progress="plain" and True maps to progress="auto".	None

Returns:

Type	Description
tuple[str, bool]	A tuple containing: - The file path where files were downloaded/cached - Boolean indicating if files were retrieved from cache (True) or downloaded (False)

Examples:

Download new files or use cache:

>>> client = SoftManager()
>>> path, cached = client.downloads(eos_obj, "/tmp/downloads")
>>> if cached:
...     print("Files retrieved from cache")

Force re-download even if cached:

>>> client = SoftManager(force_download=True)
>>> path, cached = client.downloads(eos_obj, "/tmp/downloads")
>>> cached  # Will be False
False

Source code in eos_downloader/logics/download.py

def downloads(
    self,
    object_arista: eos_downloader.logics.arista_xml_server.AristaXmlObjects,
    file_path: str,
    *,
    progress: ProgressMode = "auto",
    rich_interface: Optional[bool] = None,
) -> tuple[str, bool]:
    """
    Downloads files from Arista EOS server with caching support.

    Downloads the EOS image and optional md5/sha512 files based on the
    provided EOS XML object. Each file is downloaded to the specified path
    with appropriate filenames. Uses cache to skip already downloaded files
    unless force_download is enabled. The files of a single download are
    fetched concurrently and rendered in one shared progress display.

    Parameters
    ----------
    object_arista : eos_downloader.logics.arista_xml_server.AristaXmlObjects
        Object containing EOS image and hash file URLs.
    file_path : str
        Directory path where files should be downloaded.
    progress : ProgressMode, optional
        Progress rendering mode: ``"auto"`` (default), ``"rich"``,
        ``"plain"`` or ``"none"``.
    rich_interface : bool, optional
        Deprecated. Use ``progress`` instead. ``False`` maps to
        ``progress="plain"`` and ``True`` maps to ``progress="auto"``.

    Returns
    -------
    tuple[str, bool]
        A tuple containing:
        - The file path where files were downloaded/cached
        - Boolean indicating if files were retrieved from cache (True) or downloaded (False)

    Examples
    --------
    Download new files or use cache:

    >>> client = SoftManager()
    >>> path, cached = client.downloads(eos_obj, "/tmp/downloads")
    >>> if cached:
    ...     print("Files retrieved from cache")

    Force re-download even if cached:

    >>> client = SoftManager(force_download=True)
    >>> path, cached = client.downloads(eos_obj, "/tmp/downloads")
    >>> cached  # Will be False
    False
    """
    mode = _resolve_progress_mode(progress, rich_interface)

    logger.info(
        f"Processing files for {object_arista.version} "
        f"(force_download={self.force_download})"
    )

    if len(object_arista.urls) == 0:
        logger.error(
            f"No URLs found for download of version {object_arista.version}. "
            f"The requested version or image type may not exist on Arista servers."
        )
        raise ValueError(
            f"Filename not found for version {object_arista.version}. "
            f"Please verify that this version exists and is available for your account."
        )

    # Track if all files were retrieved from cache
    all_files_cached = True
    # Files that actually need downloading, streamed together under one display.
    to_download: List[DownloadItem] = []

    for file_type, url in sorted(object_arista.urls.items(), reverse=True):
        logger.debug(f"Processing {file_type} from {url}")
        if file_type == "image":
            filename = object_arista.filename
            self.file["name"] = filename
        else:
            filename = object_arista.hash_filename()
            self.file[file_type] = filename
        if url is None:
            logger.error(f"URL not found for {file_type}")
            raise ValueError(f"URL not found for {file_type}")
        if filename is None:
            logger.error(f"Filename not found for {file_type}")
            raise ValueError(f"Filename not found for {file_type}")

        full_path = Path(file_path) / filename
        file_was_cached = full_path.exists() and not self.force_download

        if self.dry_run:
            if file_was_cached:
                logger.info(f"[DRY-RUN] Would use cached file: {filename}")
            else:
                logger.info(
                    f"[DRY-RUN] Would download file {filename} "
                    f"for version {object_arista.version}"
                )
                all_files_cached = False
            continue

        if file_was_cached:
            logger.info(f"Using cached file: {full_path}")
        else:
            all_files_cached = False
            to_download.append((url, str(full_path), filename))

    if to_download:
        reporter = resolve_reporter(
            mode, self.console, title=f"Downloading {object_arista.version}"
        )
        download_files_concurrently(to_download, reporter)

    return file_path, all_files_cached

import_docker

import_docker(
    local_file_path: str,
    docker_name: str = "arista/ceos",
    docker_tag: str = "latest",
    force: bool = False,
) -> bool

Import local file into Docker with caching support.

Parameters:

Name	Type	Description	Default
local_file_path	str	Path to the local file to import	required
docker_name	str	Docker image name, by default “arista/ceos”	'arista/ceos'
docker_tag	str	Docker image tag, by default “latest”	'latest'
force	bool	If True, import even if image:tag already exists. Defaults to False.	False

Returns:

Type	Description
bool	True if image was retrieved from cache (already exists), False if image was imported

Raises:

Type	Description
FileNotFoundError	If the local file does not exist

Examples:

>>> manager = SoftManager()
>>> was_cached = manager.import_docker(
...     local_file_path="/downloads/cEOS-4.29.3M.tar.xz",
...     docker_name="arista/ceos",
...     docker_tag="4.29.3M"
... )
>>> if was_cached:
...     print("Image already in cache")

Source code in eos_downloader/logics/download.py

def import_docker(
    self,
    local_file_path: str,
    docker_name: str = "arista/ceos",
    docker_tag: str = "latest",
    force: bool = False,
) -> bool:
    """
    Import local file into Docker with caching support.

    Parameters
    ----------
    local_file_path : str
        Path to the local file to import
    docker_name : str, optional
        Docker image name, by default "arista/ceos"
    docker_tag : str, optional
        Docker image tag, by default "latest"
    force : bool, optional
        If True, import even if image:tag already exists.
        Defaults to False.

    Returns
    -------
    bool
        True if image was retrieved from cache (already exists),
        False if image was imported

    Raises
    ------
    FileNotFoundError
        If the local file does not exist

    Examples
    --------
    >>> manager = SoftManager()
    >>> was_cached = manager.import_docker(
    ...     local_file_path="/downloads/cEOS-4.29.3M.tar.xz",
    ...     docker_name="arista/ceos",
    ...     docker_tag="4.29.3M"
    ... )
    >>> if was_cached:
    ...     print("Image already in cache")
    """
    # Check if file exists
    if not os.path.exists(local_file_path):
        raise FileNotFoundError(f"File {local_file_path} not found")

    # Check cache unless force flag is set
    if not force and not self.force_download:
        if self._docker_image_exists(docker_name, docker_tag):
            logger.info(
                f"Docker image {docker_name}:{docker_tag} already "  # noqa: E231
                f"exists locally. Use --force to re-import."
            )
            return True

    # Log import action
    logger.info(
        f"{'[DRY-RUN] Would import' if self.dry_run else 'Importing'} "
        f"{docker_name}:{docker_tag}"  # noqa: E231
    )

    # Handle dry-run mode
    if self.dry_run:
        return False

    # Check if docker is available
    docker_path = shutil.which("docker")
    if not docker_path:
        raise FileNotFoundError("Docker binary not found in PATH")

    # Proceed with import using subprocess.run for security (no shell injection)
    try:
        import_docker_archive(
            docker_path=docker_path,
            local_file_path=str(local_file_path),
            docker_name=docker_name,
            docker_tag=docker_tag,
        )
        logger.info(
            f"Docker image {docker_name}:{docker_tag} "  # noqa: E231
            f"imported successfully"
        )
        return False  # Image was imported (not from cache)
    except subprocess.CalledProcessError as e:
        logger.error(f"Error importing docker image: {e.stderr}")
        raise RuntimeError(f"Docker import failed: {e.stderr}") from e

provision_eve

provision_eve(
    object_arista: EosXmlObject,
    noztp: bool = False,
    progress: ProgressMode = "auto",
) -> None

Provisions EVE-NG with the specified Arista EOS object.

Parameters:

Name	Type	Description	Default
object_arista	EosXmlObject	The Arista EOS object containing version, filename, and URLs.	required
noztp	bool	If True, disables ZTP (Zero Touch Provisioning). Defaults to False.	False
progress	ProgressMode	Progress rendering mode: "auto" (default), "rich", "plain" or "none".	'auto'

Raises:

Type	Description
ValueError	If no URLs are found for download or if a URL or filename is None.

Returns:

Type	Description
None

Source code in eos_downloader/logics/download.py

def provision_eve(
    self,
    object_arista: eos_downloader.logics.arista_xml_server.EosXmlObject,
    noztp: bool = False,
    progress: ProgressMode = "auto",
) -> None:
    """
    Provisions EVE-NG with the specified Arista EOS object.

    Parameters
    ----------
    object_arista : eos_downloader.logics.arista_xml_server.EosXmlObject
        The Arista EOS object containing version, filename, and URLs.
    noztp : bool, optional
        If True, disables ZTP (Zero Touch Provisioning). Defaults to False.
    progress : ProgressMode, optional
        Progress rendering mode: ``"auto"`` (default), ``"rich"``,
        ``"plain"`` or ``"none"``.

    Raises
    ------
    ValueError
        If no URLs are found for download or if a URL or filename is None.

    Returns
    -------
    None
    """

    # EVE-NG provisioning page for vEOS
    # https://www.eve-ng.net/index.php/documentation/howtos/howto-add-arista-veos/

    logger.info(
        f"Provisioning EVE-NG with {object_arista.version} / {object_arista.filename}"
    )

    file_path = f"{eos_downloader.defaults.EVE_QEMU_FOLDER_PATH}/veos-{object_arista.version}"

    filename: Union[str, None] = None
    eos_filename = object_arista.filename

    if len(object_arista.urls) == 0:
        logger.error(
            f"No URLs found for download of version {object_arista.version}. "
            f"The requested version or image type may not exist on Arista servers."
        )
        raise ValueError(
            f"Filename not found for version {object_arista.version}. "
            f"Please verify that this version exists and is available for your account."
        )

    for file_type, url in sorted(object_arista.urls.items(), reverse=True):
        logger.debug(f"Downloading {file_type} from {url}")
        if file_type == "image":
            fname = object_arista.filename
            if fname is not None:
                filename = fname
                if noztp:
                    filename = f"{os.path.splitext(fname)[0]}-noztp{os.path.splitext(fname)[1]}"
                eos_filename = filename
                logger.debug(f"filename is {filename}")
                self.file["name"] = filename
        else:
            filename = object_arista.hash_filename()
            if filename is not None:
                self.file[file_type] = filename
        if url is None:
            logger.error(f"URL not found for {file_type}")
            raise ValueError(f"URL not found for {file_type}")
        if filename is None:
            logger.error(f"Filename not found for {file_type}")
            raise ValueError(f"Filename not found for {file_type}")

        if not os.path.exists(file_path):
            logger.warning(f"creating folder on eve-ng server: {file_path}")
            self._create_destination_folder(path=file_path)

        logger.debug(
            f"downloading file {filename} for version {object_arista.version}"
        )
        self.download_file(url, file_path, filename, progress=progress)

    # Convert to QCOW2 format
    if eos_filename is None:
        raise ValueError("EOS filename not found for QCOW2 conversion")
    vmdk_path = os.path.join(file_path, eos_filename)
    qcow2_path = os.path.join(file_path, "hda.qcow2")

    if not self.dry_run:
        qemu_img_path = shutil.which("qemu-img")
        if not qemu_img_path:
            raise FileNotFoundError("qemu-img binary not found in PATH")
        convert_vmdk_to_qcow2(
            qemu_img_path=qemu_img_path,
            vmdk_path=vmdk_path,
            qcow2_path=qcow2_path,
        )
    else:
        logger.info(
            f"{'[DRY-RUN] Would convert' if self.dry_run else 'Converting'} VMDK to QCOW2 format: {vmdk_path} to {qcow2_path} "
        )

    logger.info("Applying unl_wrapper to fix permissions")
    if not self.dry_run:
        fix_eve_permissions(Path("/opt/unetlab/wrappers/unl_wrapper"))
    else:
        logger.info("[DRY-RUN] Would execute unl_wrapper to fix permissions")