Source code for matador.db.changes

# coding: utf-8
# Distributed under the terms of the MIT License.

""" This file implements an interface for querying the
__changelog_<collection> collections to allow for display and reversion
of particular database changes.


from matador.db import make_connection_to_collection
from matador.utils.print_utils import print_warning, print_notify

[docs]class DatabaseChanges: """ Class to view and undo particular database changesets. """ def __init__(self, collection_name: str, changeset_ind=0, action='view', override=False, mongo_settings=None): """ Parse arguments and run changes interface. Parameters: collection_name (str): the base collection name to act upon Keyword arguments: changset_ind (int): the number of the changset to act upon (1 is oldest) action (str): either 'view' or 'undo' override (bool): override all options to positive answers for testing mongo_settings (dict): dictionary of already-sources mongo settings """ self.changelog_name = '__changelog_{}'.format(collection_name) _, _, self.collections = make_connection_to_collection(self.changelog_name, allow_changelog=True, override=override, mongo_settings=mongo_settings) self.repo = [self.collections[key] for key in self.collections][0] curs = list(self.repo.find()) if not curs: exit('No changesets found for {}'.format(collection_name)) # if no changeset specified, print summary if changeset_ind == 0: self.print_change_summary(curs) elif changeset_ind > len(curs): exit('No changeset {} found for collection called "{}".'.format(changeset_ind, collection_name)) # otherwise, try to act on particular changeset elif changeset_ind <= len(curs): self.change = curs[changeset_ind-1] self.view_changeset(self.change, changeset_ind-1) if action == 'undo': count = curs[changeset_ind-1]['count'] print_warning('An attempt will now be made to remove {} structures from {}.' .format(count, collection_name)) print_notify('Are you sure you want to do that? (y/n)') if override: response = 'y' else: response = input() if response.lower() == 'y': print_notify('You don\'t have any doubts at all? (y/n)') if override: next_response = 'n' else: next_response = input() if next_response.lower() == 'n': print('You\'re the boss, deleting structures now...') else: exit('As I thought...') else: exit() # proceed with deletion _, _, collections = make_connection_to_collection(collection_name, allow_changelog=False, override=override) collection_to_delete_from = [collections[key] for key in collections][0] result = collection_to_delete_from.delete_many({'_id': {'$in': self.change['id_list']}}) print('Deleted {}/{} successfully.'.format(result.deleted_count, self.change['count'])) print('Tidying up changelog database...') self.repo.delete_one({'_id': self.change['_id']}) if not self.repo.find_one(): print('No structures left remaining, deleting database...') collection_to_delete_from.drop() self.repo.drop() print('Success!')
[docs] @staticmethod def view_changeset(changeset, index): """ Prints all details about a particular changeset. Parameters: changeset (dict): changeset stored in changelog database index (int): changeset index """ print('Files added by changeset:') for src in changeset['src_list']: print('(+) {}'.format(src)) print('({}) {} {:>7d} additions'.format(index+1, changeset['date'], changeset['count']))
[docs] @staticmethod def print_change_summary(curs): """ Prints a summary of changes. Parameters: curs (list): cursor from changelog database """ for index, change in enumerate(curs): print('({}) {} {:>7d} additions'.format(index+1, change['date'], change['count']))