Utils | DatasetsForecast

Utility functions for datasetsforecast

download_file

download_file(directory, source_url, decompress=False, filename=None, max_retries=3)

Download data from source_url inside directory.

Parameters:

Name	Type	Description	Default
directory	(str, Path)	Custom directory where data will be downloaded.	required
source_url	str	URL where data is hosted.	required
decompress	bool	Whether to decompress downloaded file. Default False.	False
filename	str	Override filename for the downloaded file. If None, the filename is derived from the URL.	None
max_retries	int	Maximum number of retry attempts on transient errors.	3

extract_file

extract_file(filepath, directory)

async_download_files

async_download_files(path, urls)

Asynchronously download files from urls inside path.

Parameters:

Name	Type	Description	Default
path	(str, Path)	Directory where files will be downloaded.	required
urls	Iterable[str]	Iterable of URLs to download.	required

Example:

import os
import tempfile

import requests
gh_url = 'https://api.github.com/repos/Nixtla/datasetsforecast/contents/'
base_url = 'https://raw.githubusercontent.com/Nixtla/datasetsforecast/main'

headers = {}
gh_token = os.getenv('GITHUB_TOKEN')
if gh_token is not None:
    headers = {'Authorization': f'Bearer: {gh_token}'}
resp = requests.get(gh_url, headers=headers)
if resp.status_code != 200:
    raise Exception(resp.text)
urls = [f'{base_url}/{e["path"]}' for e in resp.json() if e['type'] == 'file']
with tempfile.TemporaryDirectory() as tmp:
    tmp = Path(tmp)
    await async_download_files(tmp, urls)
    files = list(tmp.iterdir())
    assert len(files) == len(urls)

download_files

download_files(directory, urls)

Download files from urls inside directory.

Parameters:

Name	Type	Description	Default
directory	(str, Path)	Directory where files will be downloaded.	required
urls	Iterable[str]	Iterable of URLs to download.	required

Example:

with tempfile.TemporaryDirectory() as tmp:
    tmp = Path(tmp)
    fname = tmp / 'script.py'
    fname.write_text(f'''
from datasetsforecast.utils import download_files

download_files('{tmp.as_posix()}', {urls})
    ''')
    !python {fname}
    fname.unlink()
    files = list(tmp.iterdir())
    assert len(files) == len(urls)