def replicateDatabases()

in tools/db/replicateDbs.py [0:0]

• 45 lines of code
• 25 McCabe index (conditional complexity)

def replicateDatabases(args):
    """Replicate databases."""
    sourceDb = couchdb.client.Server(args.sourceDbUrl)
    targetDb = couchdb.client.Server(args.targetDbUrl)

    excludedDatabases = args.exclude.split(",")
    excludedBaseNames = [x for x in args.excludeBaseName.split(",") if x != ""]

    # Create _replicator DB if it does not exist yet.
    if "_replicator" not in sourceDb:
        sourceDb.create("_replicator")

    replicator = sourceDb["_replicator"]

    now = int(time.time())
    backupPrefix = "backup_%d_" % now

    def isExcluded(dbName):
        dbNameWithoutPrefix = dbName.replace(args.dbPrefix, "", 1)
        # is the databaseName is in the list of excluded database
        isNameExcluded = dbNameWithoutPrefix in excludedDatabases
        # if one of the basenames matches, the database is excluded
        isBaseNameExcluded = functools.reduce(lambda x, y: x or y, [dbNameWithoutPrefix.startswith(en) for en in excludedBaseNames], False)
        return isNameExcluded or isBaseNameExcluded

    # Create backup of all databases with given prefix
    print("----- Create backups -----")
    for db in [dbName for dbName in sourceDb if dbName.startswith(args.dbPrefix) and not isExcluded(dbName)]:
        backupDb = backupPrefix + db if not args.continuous else 'continuous_' + db
        replicateDesignDocument = {
            "_id": backupDb,
            "source": args.sourceDbUrl + "/" + db,
            "target": args.targetDbUrl + "/" + backupDb,
            "create_target": True,
            "continuous": args.continuous,
        }
        print("create backup: %s" % backupDb)

        filterName = "snapshotFilters"
        filterDesignDocument = sourceDb[db].get("_design/%s" % filterName)
        if not args.continuous and filterDesignDocument:
            replicateDesignDocument["filter"] = "%s/withoutDeletedAndDesignDocuments" % filterName
        replicator.save(replicateDesignDocument)

    def isBackupDb(dbName):
        return re.match("^backup_\d+_" + args.dbPrefix, dbName)

    def extractTimestamp(dbName):
        return int(dbName.split("_")[1])

    def isExpired(timestamp):
        return now - args.expires > timestamp

    # Delete all documents in the _replicator-database of old backups to avoid that they continue after they are deprecated
    print("----- Delete backup-documents older than %d seconds -----" % args.expires)
    for doc in [doc for doc in replicator.view('_all_docs', include_docs=True) if isBackupDb(doc.id) and isExpired(extractTimestamp(doc.id))]:
        print("deleting backup document: %s" % doc.id)
        # Get again the latest version of the document to delete the right revision and avoid Conflicts
        retry(lambda: replicator.delete(replicator[doc.id]), 5)

    # Delete all backup-databases, that are older than specified
    print("----- Delete backups older than %d seconds -----" % args.expires)
    for db in [db for db in targetDb if isBackupDb(db) and isExpired(extractTimestamp(db))]:
        print("deleting backup: %s" % db)
        targetDb.delete(db)