Replicate design docs for all databases in a Cloudant account

design_doc_replication_backup.py

#!/usr/bin/env python

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

#
# Summary
# =======
# This script iterates all user databases in a Cloudant account and replicates design docs to a
# new database with the suffix _bkup_YYYYMMDD_HHMMSS
#
# Usage
# =====
# $ python design_doc_replication_backup.py
# 
# Enter your Cloudant account username and password when prompted.
#

import requests, json, getpass, sys, time

sys.stdout.write('Enter your Cloudant username: ')
cl_username = raw_input()
cl_password = getpass.getpass('Enter your Cloudant password: ')

# remove domain from username if entered by the user
cl_username = cl_username.lower().replace('.cloudant.com', '')

url = 'https://{0}.cloudant.com'.format(cl_username)

AUTH = (cl_username, cl_password)

print('*' * 80)
print('Backing up design docs for {0}'.format(cl_username))
print('*' * 80)

# get a list of all databases for the cloudant user by making a call against the _all_dbs endpoint.
# For more information, see: https://docs.cloudant.com/database.html#get-databases
response = requests.get(url + '/_all_dbs', auth=AUTH)
response.raise_for_status()

all_dbs = json.loads(response.text)

# exclude these dbs from backup
exclude_db_names = ['_replicator', '_warehouser', '_users']

# iterate through the list of databases
for db_name in all_dbs:

    sys.stdout.write('DB: {0} '.format(db_name))

    if (db_name in exclude_db_names or '_bkup_' in db_name):
        print("Skipped database")
        continue

    bkup_date = time.strftime('%Y%m%d_%H%M%S')

    # get all the design documents for the database by filtering the _all_docs endpoint.
    # For more information, see: https://docs.cloudant.com/database.html#get-documents
    ddocs_url = '{0}/{1}/_all_docs?startkey=%22_design%22&endkey=%22_design0%22'.format(url, db_name)

    # make a request to get a list of design docs
    all_ddocs = json.loads(requests.get(ddocs_url, auth=AUTH).text)

    # if there is some design docs, return the ids
    all_ddoc_ids = [ ddoc['id'] for ddoc in all_ddocs['rows'] if 'id' in ddoc ]

    replicate = {
        "create_target"  : True,
        "source"         : "https://{0}:{1}@{0}.cloudant.com/{2}".format(cl_username, cl_password, db_name),
        "target"         : "https://{0}:{1}@{0}.cloudant.com/{2}_bkup_{3}".format(cl_username, cl_password, db_name, bkup_date),
        "continuous"     : False,
        # only replicate design docs
        "doc_ids"        : all_ddoc_ids
    }

    # make a synchronous request to the _replicate endpoint
    # For more infomration, see: https://docs.cloudant.com/replication.html#the-/_replicate-endpoint
    repl_response = requests.post(
                        url + '/_replicate',
                        auth = AUTH,
                        data = json.dumps(replicate),
                        headers = { 'Content-Type' : 'application/json' }
                        )

    # uh oh - something went wrong
    if repl_response.status_code != requests.codes.ok:
        print("Error submitting replication: " + repl_response.content)
        continue

    repl_response_obj = json.loads(repl_response.content)

    # first check if no changes exist in the database
    if 'no_changes' in repl_response_obj:
        print("Backed up - no_changes found for {0}_bkup_{1}".format(db_name, bkup_date))
        continue
    else:
        doc_write_failures = repl_response_obj.get('doc_write_failures', 0)

        if doc_write_failures > 0:
            print("Doc write failure " + repl_response.content)
        else:
            docs_written = repl_response_obj.get('docs_written', 0)
            print("Backed up - {0} doc(s) written to {1}_bkup_{2}".format(docs_written, db_name, bkup_date))