How to retrieve files from the platform

How to retrieve files from the platform

In this guide, you will learn how to retrieve files from Energyworx's file storage using the APIv1.5. To authenticate, you use a Bearer access token — see Automated API Access for how to create client credentials and exchange them for an access token.

Information you need

We have provided an example script below, to get you started. However, to further enable you, we will cover the script variables so you are enabled to customize it.

• project : the environment you will be downloading the files from- mind, this is not the URL or namespace, but the project ID specified in GCP.
• namespace****_id : the namespace id where the files are located, e.g. enrx_org_001
• access_token : the Bearer access token obtained with your client credentials
• download_destination_folder : the folder where you want to store the file(s). If it does not exist, it will be created.
• params : this enables you to set search parameters for the files you are targeting. All available parameters can be found in the API documentation. The most relevant ones are filename and tags
• limit : number of files to download that match the search query (params)

Example Script

    from pathlib import Path
    import requests
    import ssl
    import typing
    import urllib.request
    from urllib.parse import unquote, urlparse
    
    import numpy as np
    from tqdm import tqdm
    
    
    # Variables
    project: str = "PROJECT ID"
    namespace_id: str = "NAMESPACE ID"
    access_token: str = "ACCESS TOKEN"
    download_destination_folder: str = "DESTINATION FOLDER"
    
    params: dict = {'filename': 'Actual.csv',
              'tags': ['month:april']}
    limit: int = 1
    
    
    def get_client_session(project: str, namespace_id: str):
        headers = {
            "Accept": "application/json",
            "X-NAMESPACE": namespace_id,
            "Authorization": f"Bearer {access_token}"
        }
        session = requests.Session()
        session.headers.update(headers)
    
        global base_url
        base_url = f"https://api.{project}.energyworx.net/ewx/v1/"
    
        return session
    
    
    # Internal variables
    global base_url
    session = get_client_session(project, namespace_id)
    context = ssl.SSLContext(protocol=ssl.PROTOCOL_TLS)
    
    
    def iter_all_pages(url: str, static_params: dict, limit: int | None = None) -> typing.Iterator:
        """
        Retrieve all pages in a collection.
    
        """
    
        # Loop over all pages in this collection and yield everything
        item_limit = np.infty if limit is None else limit
        try:
            more = True
            page = None
            while more:
                limit_size = min(100, item_limit)
                params = {"pageToken": page, "limit": limit_size}
                params.update(static_params)
                data = session.get(url, params=params).json()
                yield data
                page = data.get("pageToken") or data.get("nextPageToken")
                item_limit -= limit_size
                more = (data.get("more") and item_limit)
        except Exception:
            raise
    
    
    def get_all_file_entries():
        for page in iter_all_pages(f"{base_url}storage/files/search", params, limit):
            yield from page.get('items', [])
    
    
    def get_file_urls():
        return [x['signedUrl'] for x in get_all_file_entries()]
    
    
    def download_files(file_urls: list[str], dest_dir: str):
        # Create destination folder
        dest_dir = Path(dest_dir)
        dest_dir.mkdir(parents=True, exist_ok=True)
    
        # Download every single file
        print(f"Destination directory set to {dest_dir.name}.")
        pbar = tqdm(file_urls, desc='Downloading files...', dynamic_ncols=True)
        for file_url in pbar:
            file_name = unquote(Path(urlparse(file_url).path).name)
            pbar.set_postfix_str(file_name)
            with urllib.request.urlopen(file_url, context=context) as file:
                file_contents = file.read().decode('ascii')
            with open(dest_dir / (file_name.replace(':', '_')), 'w') as file:
                file.write(file_contents)
    
    
    def main():
        download_files(get_file_urls(), download_destination_folder)
    
    
    if __name__ == '__main__':
        main()