Download Management

download

ObjectDownloader class to manage file downloads with an option to use rich interface.

This class provides methods to download files from URLs with progress tracking using either tqdm or rich interface. It supports both raw downloads and enhanced visual feedback during the download process.

Functions:

Name	Description
download_file	Downloads a file from the given URL to the specified path with optional rich interface.
_download_file_raw	Static method that performs the actual file download with tqdm progress bar.

Attributes:

Name	Type	Description
None

Example

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

SoftManager

SoftManager(
    dry_run: bool = False, force_download: bool = False
)

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

Source code in eos_downloader/logics/download.py

def __init__(self, dry_run: bool = False, force_download: bool = False) -> 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
    """
    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
    logging.info(
        "SoftManager initialized%s%s",
        " 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
    """
    logging.info(f"Checking checksum for {self.file['name']} using {check_type}")

    if self.dry_run:
        logging.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"]

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

        if file_name is None or hash512sum is None:
            logging.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:
            logging.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):
            logging.error(
                f"Checksum failed for {self.file['name']}: expected {hash_expected}"
            )

            raise ValueError("Incorrect checksum")

        return True

    logging.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,
    *,
    rich_interface: bool = True,
    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
rich_interface	bool	Whether to use rich progress bar interface. Defaults to True.	True
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,
    *,
    rich_interface: bool = True,
    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.
    rich_interface : bool, optional
        Whether to use rich progress bar interface. Defaults to True.
    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'
    """
    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():
            logging.info(f"Using cached file: {full_path}")
            return str(full_path)

    # Log download action
    logging.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:
        if not rich_interface:
            return self._download_file_raw(
                url=url, file_path=os.path.join(file_path, filename)
            )
        rich_downloader = eos_downloader.helpers.DownloadProgressBar()
        rich_downloader.download(urls=[url], dest_dir=file_path)
        return os.path.join(file_path, filename)

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

downloads

downloads(
    object_arista: AristaXmlObjects,
    file_path: str,
    rich_interface: bool = True,
) -> 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.

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
rich_interface	bool	Whether to use rich console output. Defaults to True.	True

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,
    rich_interface: bool = True,
) -> 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.

    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.
    rich_interface : bool, optional
        Whether to use rich console output. Defaults to True.

    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
    """
    logging.info(
        f"Processing files for {object_arista.version} "
        f"(force_download={self.force_download})"
    )

    if len(object_arista.urls) == 0:
        logging.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

    for file_type, url in sorted(object_arista.urls.items(), reverse=True):
        logging.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:
            logging.error(f"URL not found for {file_type}")
            raise ValueError(f"URL not found for {file_type}")
        if filename is None:
            logging.error(f"Filename not found for {file_type}")
            raise ValueError(f"Filename not found for {file_type}")
        if not self.dry_run:
            # Check if file is already cached before download
            full_path = Path(file_path) / filename
            file_was_cached = full_path.exists() and not self.force_download

            # download_file will check cache automatically
            # unless self.force_download is True
            self.download_file(
                url,
                file_path,
                filename,
                rich_interface=rich_interface,
                force=self.force_download,
            )

            # Update cache tracking
            if not file_was_cached:
                all_files_cached = False
        else:
            full_path = Path(file_path) / filename
            if full_path.exists() and not self.force_download:
                logging.info(f"[DRY-RUN] Would use cached file: {filename}")
            else:
                logging.info(
                    f"[DRY-RUN] Would download file {filename} "
                    f"for version {object_arista.version}"
                )
                all_files_cached = False

    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):
            logging.info(
                f"Docker image {docker_name}:{docker_tag} already "  # noqa: E231
                f"exists locally. Use --force to re-import."
            )
            return True

    # Log import action
    logging.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
    if not shutil.which("docker"):
        raise FileNotFoundError("Docker binary not found in PATH")

    # Proceed with import
    try:
        cmd = (
            f"$(which docker) import {local_file_path} "
            f"{docker_name}:{docker_tag}"  # noqa: E231
        )
        logging.debug(f"Executing: {cmd}")
        os.system(cmd)
        logging.info(
            f"Docker image {docker_name}:{docker_tag} "  # noqa: E231
            f"imported successfully"
        )
        return False  # Image was imported (not from cache)
    except Exception as e:
        logging.error(f"Error importing docker image: {e}")
        raise e

provision_eve

provision_eve(
    object_arista: EosXmlObject, noztp: bool = False
) -> 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

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,
) -> 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.

    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/

    logging.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:
        logging.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):
        logging.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
                logging.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:
            logging.error(f"URL not found for {file_type}")
            raise ValueError(f"URL not found for {file_type}")
        if filename is None:
            logging.error(f"Filename not found for {file_type}")
            raise ValueError(f"Filename not found for {file_type}")

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

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

    # Convert to QCOW2 format
    file_qcow2 = os.path.join(file_path, "hda.qcow2")

    if not self.dry_run:
        os.system(
            f"$(which qemu-img) convert -f vmdk -O qcow2 {file_path}/{eos_filename} {file_path}/{file_qcow2}"
        )
    else:
        logging.info(
            f"{'[DRY-RUN] Would convert' if self.dry_run else 'Converting'} VMDK to QCOW2 format: {file_path}/{eos_filename} to {file_qcow2} "
        )

    logging.info("Applying unl_wrapper to fix permissions")
    if not self.dry_run:
        os.system("/opt/unetlab/wrappers/unl_wrapper -a fixpermissions")
    else:
        logging.info("[DRY-RUN] Would execute unl_wrapper to fix permissions")