cdm_data_loaders.utils.file_transfer.sync_client ¶

class FileDownloader(BaseFileDownloader):
    """Synchronous downloader interface."""

    def __init__(
        self,
        client: httpx.Client | None = None,
        max_attempts: int = RETRY_DEFAULTS["max_attempts"],
        min_backoff: int = RETRY_DEFAULTS["min_backoff"],
        max_backoff: int = RETRY_DEFAULTS["max_backoff"],
        chunk_size: int = 8192,
    ) -> None:
        """Initialise a synchronous download client.

        :param client: an httpx.Client object, defaults to None
        :type client: httpx.Client | None, optional
        :param max_attempts: how many times to retry the request defaults to 5
        :type max_attempts: int, optional
        :param min_backoff: minimum backoff for retries, defaults to 1
        :type min_backoff: int, optional
        :param max_backoff: maximum backoff for retries, defaults to 30
        :type max_backoff: int, optional
        :param chunk_size: chunk size for the downloader, defaults to 8192
        :type chunk_size: int, optional
        """
        super().__init__(client or get_httpx_client(), chunk_size=chunk_size)

        self._retry = retry(
            **DownloadCore.retry_kwargs(max_attempts, min_backoff, max_backoff),
            before_sleep=before_sleep_log(logger, WARNING),
        )

    def download(
        self,
        url: str,
        destination: str | Path,
        expected_checksum: str | None = None,
        checksum_fn: str | None = None,
        extra_headers: dict[str, Any] | None = None,
    ) -> Path | None:
        """Download a file from ``url`` to ``destination`` on disk.

        Returns the file path if successful; None otherwise.
        304 / Not Modified responses return None.

        :param url: URL to download from
        :type url: str
        :param destination: where to save the file to
        :type destination: str | Path
        :param expected_checksum: expected checksum (if known), defaults to None
        :type expected_checksum: str | None, optional
        :param checksum_fn: function used for calculating the checksum. Set to "sha256" if not supplied.
        :type checksum_fn: str, optional
        :param extra_headers: allow extra headers to be passed, defaults to None
        :type extra_headers: dict[str, Any] | None, optional
        :return: either the path to the downloaded file or None if no file was downloaded
        :rtype: Path | None
        """
        destination, checksum_fn, extra_headers = DownloadCore.validate_args(destination, checksum_fn, extra_headers)

        @self._retry
        def _once() -> Path | None:
            return self._download_once(url, destination, expected_checksum, checksum_fn, extra_headers)

        return _once()

    def _download_once(
        self,
        url: str,
        destination: Path,
        expected_checksum: str | None,
        checksum_fn: str,
        extra_headers: dict[str, Any],
    ) -> Path | None:
        """Core synchronous download function.

        :param url: URL to download from
        :type url: str
        :param destination: where to save the file to
        :type destination: Path
        :param expected_checksum: expected checksum (if known), defaults to None
        :type expected_checksum: str | None, optional
        :param checksum_fn: function used for calculating the checksum, defaults to "sha256"
        :type checksum_fn: str, optional
        :param extra_headers: any extra headers to pass to the request, defaults to None
        :type extra_headers: dict[str, Any] | None, optional
        :raises DownloadError: for unrecoverable download errors
        :raises ChecksumMismatchError: if the checksum does not match expected_checksum
        :return: either the path to the downloaded file or None if no file was downloaded
        :rtype: Path | None
        """
        try:
            with self.client.stream("GET", url, headers=extra_headers) as response:
                if not DownloadCore.check_should_download(response, url, logger):
                    return None

                chunks = response.iter_bytes(chunk_size=self.chunk_size)

                DownloadCore.write_and_hash(
                    url,
                    destination,
                    chunks,
                    expected_checksum=expected_checksum,
                    checksum_fn=checksum_fn,
                )

        except Exception as exc:  # noqa: BLE001
            DownloadCore.handle_download_exception(exc, url, logger)

        return DownloadCore.log_download_success(url, destination, logger)