Source code for

Contains functionality for downloading files from a URL. Intended for
downloading files from |PlasmaPy's data repository|.

import contextlib
import json
import time
import warnings
from pathlib import Path
from urllib.parse import urljoin

import requests

__all__ = ["Downloader"]

# TODO: use a config file variable to allow users to set a location
# for the data download folder?

[docs] class Downloader: """ Accesses the PlasmaPy resource files. Retrieves local paths to resource files, and downloads those files from |PlasmaPy's data repository| if they cannot be found locally. Parameters ---------- directory : `~pathlib.Path`, optional The directory into which files will be downloaded. The default is :file:`/~//.plasmapy//downloads//` validate : `bool`, default: `True` If `True`, verify that local files are up-to-date with the data repository, and use the GitHub API to verify download URLs before downloading files. If `False`, return any matching local file without verification and, if a local file cannot be found, attempt to download from the repository without validation. api_token : `str`, optional A GitHub authorization token that, if provided, will be used for queries to the GitHub API. If none is provided, public API calls will be used. """ # URL for the PlasmaPy-data repository through the GitHub API _API_BASE_URL = "" # Name for the local file that stores the SHA hashes of downloaded files # and information about the SHA on the server _blob_file_name = "RESOURCE_BLOB_SHA.json" # Base URL for RAW files _RAW_BASE_URL = "" def __init__( self, directory: Path | None = None, validate: bool = True, api_token: str | None = None, ): if directory is None: # No test coverage for default directory, since pytest always # saves into a temporary directory self._download_directory = ( Path.home() / ".plasmapy" / "downloads" ) # coverage: ignore else: self._download_directory = Path(directory) self._validate = validate self._api_token = api_token # Flag to record whether the blob file has been updated from the repo # by this instantiation of the class. Once the file has been updated # once, we won't update it again to limit API calls. self._updated_blob_file_from_repo = False self._download_directory.mkdir(parents=True, exist_ok=True) # Path to the local SHA blob file self._blob_file = Path(self._download_directory, self._blob_file_name) # Create the SHA blob file if it doesn't already exist if not self._blob_file.is_file(): self._blob_dict = {} self._write_blobfile() # Otherwise, read the SHA blob file else: self._read_blobfile() def _write_blobfile(self) -> None: """ Write the _local_blob_dict to disk. """ with"w") as f: json.dump(self._blob_dict, fp=f) def _read_blobfile(self) -> None: """ Read the _local_blob_dict from disk. """ with"r") as f: self._blob_dict = json.load(f) @property def _api_connected(self) -> bool: """ Return `True` if a connection exists to the API, otherwise `False`. """ try: # Requesting this URL does not count as an API query self._http_request("") # No testing because CI always has a connection to the API except requests.ConnectionError: # coverage: ignore return False return True @property def _api_usage(self) -> tuple[int, int]: """ Return the API call limit and the number currently used from this IP. """ # Ensure that the GitHub API is not rate limited reply = self._http_request("") info = reply.json() rate_info = info["resources"]["core"] limit = int(rate_info["limit"]) used = int(rate_info["used"]) return limit, used @property def _api_is_rate_limited(self) -> bool: """ Whether or not the API is currently rate limited. """ limit, used = self._api_usage return used >= limit @property def _do_validation(self) -> bool: """ Determine whether or not to enforce validation using the GitHub API. """ return self._validate and self._api_connected and not self._api_is_rate_limited def _update_repo_blob_dict(self) -> None: """ Update the blob file with a call to the repository. Raises ------ ValueError If the URL does not return the expected JSON file with the expected keys. Returns ------- repo_blob_dict : dict Dictionary with filenames as keys. Each item is another entry with keys ``"sha"`` and ``"download_url"``. """ # If the current blob file has been updated in the past 5 minutes, # don't bother doing it again # Ignore in tests, as this won't happen in CI with contextlib.suppress(KeyError): # If the _timestamp key hasn't been set yet, the blob file has # never been updated before if time.time() - self._blob_dict["_timestamp"] < 300: return None # coverage : ignore # If this instance of Downloader has already updated from the API once, # don't do it again. Almost certainly nothing has changed! # Not tested, since CI never waits >5 min with the same Downloader # instantiated. if self._updated_blob_file_from_repo: # coverage: ignore return None reply = self._http_request(self._API_BASE_URL) # Extract the SHA hash and the download URL from the response # Extract contents to JSON # Not tested, since any URL on the GitHub API that doesn't raise a 404 error # should return a JSON try: # coverage: ignore info = reply.json() except requests.exceptions.JSONDecodeError as err: # coverage: ignore warnings.warn( "URL did not return the expected JSON file: " f"{self._API_BASE_URL}. " f"Response content: {reply.content}. Exception: {err}" ) self._validate = False return None for item in info: try: filename = item["name"] repo_sha = item["sha"] download_url = item["download_url"] # Not tested, since any URL on the GitHub API that doesn't return a 404 # should be a JSON with these keys except (KeyError, TypeError) as err: # coverage: ignore warnings.warn( f"URL {self._API_BASE_URL} returned JSON file, " "missing expected keys 'sha' and 'download_url`." f" JSON contents: {info}. Exception: {err}" ) filename = None repo_sha = None download_url = None if filename is not None: self._update_blob_entry( filename, repo_sha=repo_sha, download_url=download_url ) # Save the current epoch time in the blob file as a record of when # it was updated self._blob_dict["_timestamp"] = time.time() # At the end, write back to the blobfile self._write_blobfile() # The blob file has been updated: set this flag so we won't do it # again on this instance of Downloader self._updated_blob_file_from_repo = True def _update_blob_entry( self, filename: str, local_sha: str | None = None, repo_sha: str | None = None, download_url: str | None = None, ) -> None: """ Update an entry in the blobfile, or create a new one if one doesn't exist. """ if filename in self._blob_dict: if local_sha is not None: self._blob_dict[filename]["local_sha"] = local_sha if repo_sha is not None: self._blob_dict[filename]["repo_sha"] = repo_sha if download_url is not None: self._blob_dict[filename]["download_url"] = download_url else: self._blob_dict[filename] = { "local_sha": local_sha, "repo_sha": repo_sha, "download_url": download_url, } def _http_request(self, url: str) -> requests.Response: """ Issue an HTTP request to the specified URL, handling exceptions. """ # Only send GitHub api authorization if querying GitHub # auth = self._api_auth if "" in url else None headers = {"Content-Type": "application/json"} if self._api_token is not None: headers["authorization"] = f"Bearer {self._api_token}" try: reply = requests.get(url, headers=headers, timeout=10) # No test coverage for this exception since we can't test it without # severing the network connectivity in pytest except requests.ConnectionError as err: # coverage: ignore raise requests.ConnectionError( f"Unable to connect to data repository {self._API_BASE_URL}" ) from err # Extract the 'message' value if it is there # If the file does not exist on the repository, the GitHub API # will return `Not Found` in response to this but not raise a 404 error if reply.status_code == 404: raise ValueError(f"URL returned 404: {url}") return reply def _filepath(self, filename: str) -> Path: """Formats a filepath from a filename.""" return Path(self._download_directory, filename) def _download_file(self, filename: str, dl_url: str) -> Path: """ Download a file from a given URL to a specified path. Parameters ---------- filename : str Name of the file to download: determines the filepath dl_url : str URL from which to download Returns ------- filepath : str Path to the downloaded file """ # Request the contents of the file from the download URL reply = self._http_request(dl_url) filepath = self._filepath(filename) # Write the contents to file with"wb") as f: f.write(reply.content) return filepath def _get_file_without_validation(self, filename: str) -> Path: """ Return file logic without validation. Returns ------- filepath : Path Path to file. Raises ------ ValueError If the resource cannot be found locally or on the repository. """ filepath = self._filepath(filename) # If the file exists locally, return that if filepath.is_file(): return filepath # Try blindly downloading from the base URL # Note that downloading directly from the RAW url does not # require an API call. with contextlib.suppress(ValueError): dl_url = urljoin(self._RAW_BASE_URL, filename) return self._download_file(filename, dl_url) raise ValueError( "Resource could not be found locally or " "retrieved from the PlasmaPy-data repository: " f"{filename}." ) def _get_file_with_validation(self, filename: str) -> Path: """ Return file logic with validation. """ filepath = self._filepath(filename) self._update_repo_blob_dict() # Retrieve the values: try/except catches KeyError both for # key `filename` and `local_sha`/`repo_sha` try: local_sha = self._blob_dict[filename]["local_sha"] except KeyError: local_sha = None try: repo_sha = self._blob_dict[filename]["repo_sha"] except KeyError: repo_sha = None # If local sha and online sha are equal, return the local filepath if local_sha == repo_sha and local_sha is not None: return filepath # If the file is found online, try downloading from the repository elif repo_sha is not None: dl_url = self._blob_dict[filename]["download_url"] # Download the file filepath = self._download_file(filename, dl_url) # This is a verified download, so we now know the local_sha is # the same as the repo_sha self._update_blob_entry(filename, local_sha=repo_sha) self._write_blobfile() return filepath # Otherwise fall back to retrieving the file without validation else: self._validate = False warnings.warn( f"Could not retrieve file {filename} with validation: " "trying again without validation." ) return self._get_file_without_validation(filename)
[docs] def get_file(self, filename: str) -> Path: """ Returns a local path to a resource file, downloading it if necessary. Parameters ---------- filename : str The name of the file in the |PlasmaPy's data repository|. Returns ------- Path : `~pathlib.Path` The local path to the resource file. """ if self._do_validation: return self._get_file_with_validation(filename) return self._get_file_without_validation(filename)