--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/borgend/repository.py Sun Jan 28 11:54:46 2018 +0000
@@ -0,0 +1,143 @@
+#
+# Repository abstraction for queuing
+#
+
+import weakref
+import keyring
+
+from . import loggers
+from . import config
+from .scheduler import QueueThread, QueuedEvent
+
+logger=loggers.get(__name__)
+
+class FIFOEvent(QueuedEvent):
+ def __init__(self, cond, name=None):
+ self._goodtogo=False
+ super().__init__(cond, name=name)
+
+ def __lt__(self, other):
+ return False
+
+class FIFO(QueueThread):
+ def __init__(self, **kwargs):
+ super().__init__(target = self._fifo_thread, **kwargs)
+
+ def _fifo_thread(self):
+ with self._cond:
+ while not self._terminate:
+ ev=self._list
+ if ev:
+ # We can only remove ev from the list when ev.cond allows
+ with ev.cond:
+ if not ev._goodtogo:
+ ev._goodtogo=True
+ ev.cond.notifyAll()
+ self._cond.wait()
+
+ # Termination cleanup
+ ev=self._list
+ while ev:
+ # We can only remove ev from the list when ev.cond allows
+ with ev.cond:
+ ev.cond.notifyAll()
+ ev=ev.next
+
+ # cond has to be acquired on entry!
+ def queue_action(self, cond, action=lambda: (), name=None):
+ ev=FIFOEvent(cond, name=name)
+
+ with self._cond:
+ self._insert(ev)
+
+ # This will release the lock on cond, allowing queue manager (scheduler)
+ # thread to notify us if we are already to be released
+ logger.debug("%s:Queuing %s", self.name, ev.name or 'UNKNOWN')
+ ev.cond.wait()
+
+ try:
+ if ev._goodtogo:
+ logger.debug("%s:Executing %s", self.name, ev.name or 'UNKNOWN')
+ #
+ # TODO: action() has to unlink on finish; so should maybe
+ # use weak references to event.
+ # Or we have to make action take all the time, so make the
+ # stdout thread.
+ # OR: Easiest to just move finish-waiting into __launch_check
+ # instead of at the outer level of the main loop.
+ #
+ action()
+ finally:
+ with self._cond:
+ self._unlink(ev)
+ # Let _fifo_thread proceed to next action
+ self._cond.notify()
+
+ return ev._goodtogo
+
+repositories=weakref.WeakValueDictionary()
+
+class Repository(FIFO):
+ def __decode_config(self, cfg):
+ loc0='Repository %d' % self.identifier
+
+ self.repository_name=config.check_string(cfg, 'name', 'Name', loc0)
+
+ logger.debug("Configuring repository '%s'" % self.repository_name)
+
+ loc = 'Repository "%s"'
+
+ self.logger=logger.getChild(self.repository_name)
+
+ self.location=config.check_string(cfg, 'location',
+ 'Target repository location', loc)
+
+ self.borg_parameters=config.BorgParameters.from_config(cfg, loc)
+
+ self.__keychain_account=config.check_string(cfg, 'keychain_account',
+ 'Keychain account', loc,
+ default='')
+
+ self.__passphrase=None
+
+ if config.settings['extract_passphrases_at_startup']:
+ try:
+ self.extract_passphrase()
+ except Exception:
+ pass
+
+ def __init__(self, identifier, cfg):
+ self.identifier=identifier
+ self.__decode_config(cfg)
+ super().__init__(name = 'RepositoryThread %s' % self.repository_name)
+ repositories[self.repository_name]=self
+
+ def __extract_passphrase(self):
+ acc=self.__keychain_account
+ if not self.__passphrase:
+ if acc and acc!='':
+ self.logger.debug('Requesting passphrase')
+ try:
+ pw=keyring.get_password("borg-backup", acc)
+ except Exception as err:
+ self.logger.error('Failed to retrieve passphrase')
+ raise err
+ else:
+ self.logger.debug('Received passphrase')
+ self.__passphrase=pw
+ else:
+ self.__passphrase=None
+ return self.__passphrase
+
+ def launch_borg_instance(self, inst):
+ with self._cond:
+ passphrase=self.__extract_passphrase()
+ inst.launch(passphrase=passphrase)
+
+def find_repository(name):
+ if name in repositories:
+ return repositories[name]
+ else:
+ return None
+
+