Python: Download all the maps from your account

.gitignore

*.carto

download-all-maps.py

import argparse
import logging
import os
import warnings
import re
import requests
from pathlib import Path
from multiprocessing import Pool
from carto.auth import APIKeyAuthClient
from carto.visualizations import VisualizationManager
from carto.exceptions import CartoException

warnings.filterwarnings('ignore')

# Logger (better than print)
logging.basicConfig(
    level=logging.INFO,
    format=' %(asctime)s - %(levelname)s - %(message)s',
    datefmt='%I:%M:%S %p')
logger = logging.getLogger()

# set input arguments
parser = argparse.ArgumentParser(
    description='''
        Export all visualizations on your CARTO account,
        the maps are used as filenames after sanitization
        ''')

parser.add_argument('--processes', '-p', type=int, dest='processes', default=5,
                    help="Number of concurrent downloading processes, defaults to 5")

parser.add_argument('--organization', '-o', type=str, dest='organization',
                    default=os.getenv('CARTO_ORG', None),
                    help='Set the name of the organization' +
                    ' account (defaults to env variable CARTO_ORG)')

parser.add_argument('--api_url', '-u', type=str, dest='CARTO_API_URL',
                    default=os.getenv('CARTO_API_URL', ''),
                    help='Set the base URL. For example:' +
                    ' https://username.carto.com/ ' +
                    '(defaults to env variable CARTO_API_URL)')

parser.add_argument('--api_key', '-a', dest='CARTO_API_KEY',
                    default=os.getenv('CARTO_API_KEY', ''),
                    help='Api key of the account' +
                    ' (defaults to env variable CARTO_API_KEY)')


def download_map(viz):
    # import ipdb; ipdb.set_trace()
    name = viz.name
    safe_name = "".join([c for c in name if re.match(r'\w', c)])
    p = Path(f'{safe_name}.carto')

    if not p.exists():
        logger.info(f'Generating URL for {p}')
        try:
            url = viz.export()
            logger.info(f'Downloading {p}...')
            with p.open('wb') as fd:
                r = requests.get(url, allow_redirects=True, stream=True)
                for chunk in r.iter_content(chunk_size=1024):
                    fd.write(chunk)
                logger.info(f'Finished {p}')
                return {'finished': viz}
        except CartoException as e:
            logger.error('Error downloading {}: {}'.format(name, str(e)))
            return {'errored': viz}
    else:
        logger.debug(f'Skipping {p}')
        return {'skipped': viz}


if __name__ == '__main__':
    args = parser.parse_args()

    # Set authentification to CARTO
    if args.CARTO_API_URL and args.CARTO_API_KEY:
        auth_client = APIKeyAuthClient(
            args.CARTO_API_URL,
            args.CARTO_API_KEY,
            args.organization
        )
    else:
        logger.error(
            'You need to provide valid credentials, run with -h parameter for details')
        import sys
        sys.exit(1)

    logger.info('Getting all visualizations...')
    visualizations = VisualizationManager(auth_client).all()
    with Pool(args.processes) as p:
        logger.info('Starting the download...')
        results = p.map(download_map, visualizations)

    split_results = {}
    logger.info('Results:')
    for part in ['finished', 'skipped', 'errored']:
        # Unzip the parallel results into a single subset
        # so first filter them and then take the visualization out
        split_results[part] = list(
                map(lambda v: v[part],
                    filter(lambda v: part in v, results)))
        logger.info('{:3d} {}'.format(len(split_results[part]), part))

    if len(split_results['errored']) > 0:
        logger.info("\r\n".join(
            map(lambda v: v.name, split_results['errored'])))
    logger.info('Done!')