db_backup.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""db_backup.py
"""

import csv
import gzip
import json
import time
import shutil
import urllib.request

from pathlib import Path
from os import environ as env
from dotenv import load_dotenv

from auth0.v3.authentication import GetToken
from auth0.v3.management import Auth0

import constants


def connect_to_auth0():
    """Connect to Auth0 using Client Credentials flow. Credentials are stored
    in .env"""
    env_path = Path('.') / '.env'
    load_dotenv(dotenv_path=env_path)
    auth0_client_id = env[constants.AUTH0_CLIENT_ID]
    auth0_client_secret = env[constants.AUTH0_CLIENT_SECRET]
    auth0_domain = env[constants.AUTH0_DOMAIN]
    mgmt_api_url = 'https://'+auth0_domain+'/api/v2/'

    get_token = GetToken(auth0_domain)
    token = get_token.client_credentials(auth0_client_id,
                                         auth0_client_secret,
                                         mgmt_api_url)
    mgmt_api_token = token['access_token']
    return Auth0(auth0_domain, mgmt_api_token)


def main():
    """main"""

    auth0 = connect_to_auth0()

    export_job = {}
    export_job_id = auth0.jobs.export_users(export_job)['id']

    while auth0.jobs.get(export_job_id)['status'] != 'completed':
        time.sleep(5)

    export_job_output_dir = './output'
    export_job_output_gz_filename = 'export.csv.gz'
    export_job_output_gz_filepath = (export_job_output_dir +
                                     '/' + export_job_output_gz_filename)
    urllib.request.urlretrieve(auth0.jobs.get(export_job_id)['location'],
                               export_job_output_gz_filepath)

    export_job_output_csv_filename = 'export.csv'
    export_job_output_csv_filepath = (export_job_output_dir +
                                      '/' + export_job_output_csv_filename)
    with gzip.open(export_job_output_gz_filepath, 'rb') as gz_file:
        with open(export_job_output_csv_filepath, 'wb') as csv_file:
            shutil.copyfileobj(gz_file, csv_file)

    users = []

    with open(export_job_output_csv_filepath, 'r') as csv_file:
        csv_reader = csv.reader(csv_file, delimiter=',')
        next(csv_reader)
        for user in csv_reader:
            users.append(auth0.users.get(user[0]))
            time.sleep(1)

    print(json.dumps(users, sort_keys=True, indent=2, separators=(',', ':')))


if __name__ == '__main__':
    main()