1
0
Fork 0
mirror of synced 2024-07-13 20:50:02 -04:00

Initial commit

This commit is contained in:
Anish Athalye 2014-03-19 23:07:30 -04:00
commit 60a560e976
16 changed files with 347 additions and 0 deletions

1
.gitignore vendored Normal file
View file

@ -0,0 +1 @@
*.pyc

19
bin/dotbot Executable file
View file

@ -0,0 +1,19 @@
#!/usr/bin/env python
import sys, os
PROJECT_ROOT_DIRECTORY = os.path.dirname(
os.path.dirname(os.path.realpath(__file__)))
if os.path.exists(os.path.join(PROJECT_ROOT_DIRECTORY, 'dotbot')):
if PROJECT_ROOT_DIRECTORY not in sys.path:
sys.path.insert(0, PROJECT_ROOT_DIRECTORY)
os.putenv('PYTHONPATH', PROJECT_ROOT_DIRECTORY)
import dotbot
def main():
dotbot.cli.main()
if __name__ == '__main__':
main()

1
dotbot/__init__.py Normal file
View file

@ -0,0 +1 @@
from .cli import main

46
dotbot/cli.py Normal file
View file

@ -0,0 +1,46 @@
from argparse import ArgumentParser
from .config import ConfigReader, ReadingError
from .dispatcher import Dispatcher, DispatchError
from .messenger import Messenger
from .messenger import Level
def add_options(parser):
parser.add_argument('-Q', '--super-quiet', dest = 'super_quiet', action = 'store_true',
help = 'suppress almost all output')
parser.add_argument('-q', '--quiet', dest = 'quiet', action = 'store_true',
help = 'suppress most output')
parser.add_argument('-v', '--verbose', dest = 'verbose', action = 'store_true',
help = 'enable verbose output')
parser.add_argument('-d', '--base-directory', nargs = 1,
dest = 'base_directory', help = 'execute commands from within BASEDIR',
metavar = 'BASEDIR', required = True)
parser.add_argument('-c', '--config-file', nargs = 1, dest = 'config_file',
help = 'run commands given in CONFIGFILE', metavar = 'CONFIGFILE',
required = True)
def read_config(config_file):
reader = ConfigReader(config_file)
return reader.get_config()
def main():
log = Messenger()
try:
parser = ArgumentParser()
add_options(parser)
options = parser.parse_args()
if (options.super_quiet):
log.set_level(Level.WARNING)
if (options.quiet):
log.set_level(Level.INFO)
if (options.verbose):
log.set_level(Level.DEBUG)
tasks = read_config(options.config_file[0])
dispatcher = Dispatcher(options.base_directory[0])
success = dispatcher.dispatch(tasks)
if success:
log.info('\n==> All tasks executed successfully')
else:
raise DispatchError('\n==> Some tasks were not executed successfully')
except (ReadingError, DispatchError) as e:
log.error('%s' % e)
exit(1)

19
dotbot/config.py Normal file
View file

@ -0,0 +1,19 @@
import json
class ConfigReader(object):
def __init__(self, config_file_path):
self._config = self._read(config_file_path)
def _read(self, config_file_path):
try:
with open(config_file_path) as fin:
data = json.load(fin)
return data
except Exception:
raise ReadingError('Could not read config file')
def get_config(self):
return self._config
class ReadingError(Exception):
pass

46
dotbot/dispatcher.py Normal file
View file

@ -0,0 +1,46 @@
import os
from .executor import *
from .messenger import Messenger
class Dispatcher(object):
PLUGIN_CLASS = Executor
PLUGIN_DIR = 'dotbot/executor'
def __init__(self, base_directory):
self._log = Messenger()
self._set_base_directory(base_directory)
self._load_plugins()
def _set_base_directory(self, base_directory):
path = os.path.abspath(os.path.realpath(
os.path.expanduser(base_directory)))
if os.path.exists(path):
self._base_directory = path
else:
raise DispatchError('Nonexistant base directory')
def dispatch(self, tasks):
success = True
for task in tasks:
for action in task:
handled = False
for plugin in self._plugins:
if plugin.can_handle(action):
try:
success &= plugin.handle(action, task[action])
handled = True
except Exception:
self._log.error(
'An error was encountered while executing action %s' %
action)
if not handled:
success = False
self._log.error('Action %s not handled' % action)
return success
def _load_plugins(self):
self._plugins = [plugin(self._base_directory)
for plugin in Executor.__subclasses__()]
class DispatchError(Exception):
pass

View file

@ -0,0 +1,3 @@
from .executor import Executor
from .linker import Linker
from .commandrunner import CommandRunner

View file

@ -0,0 +1,32 @@
import os, subprocess
from . import Executor
class CommandRunner(Executor):
'''
Run arbitrary shell commands.
'''
def can_handle(self, directive):
return directive == 'shell'
def handle(self, directive, data):
if directive != 'shell':
raise ValueError('CommandRunner cannot handle directive %s' %
directive)
return self._process_commands(data)
def _process_commands(self, data):
success = True
with open(os.devnull, 'w') as devnull:
for cmd, msg in data:
self._log.lowinfo('%s [%s]' % (msg, cmd))
ret = subprocess.call(cmd, shell = True, stdout = devnull,
stderr = devnull, cwd = self._base_directory)
if ret != 0:
success = False
self._log.warning('Command [%s] failed' % cmd)
if success:
self._log.info('All commands have been executed')
else:
self._log.error('Some commands were not sucessfully executed')
return success

View file

@ -0,0 +1,24 @@
from ..messenger import Messenger
class Executor(object):
'''
Abstract base class for commands that process directives.
'''
def __init__(self, base_directory):
self._base_directory = base_directory
self._log = Messenger()
def can_handle(self, directive):
'''
Returns true if the Executor can handle the directive.
'''
raise NotImplementedError
def handle(self, directive, data):
'''
Executes the directive.
Returns true if the Executor successfully handled the directive.
'''
raise NotImplementedError

73
dotbot/executor/linker.py Normal file
View file

@ -0,0 +1,73 @@
import os
from . import Executor
class Linker(Executor):
'''
Symbolically links dotfiles.
'''
def can_handle(self, directive):
return directive == 'link'
def handle(self, directive, data):
if directive != 'link':
raise ValueError('Linker cannot handle directive %s' % directive)
return self._process_links(data)
def _process_links(self, links):
success = True
for destination, source in links.items():
success &= self._link(source, destination)
if success:
self._log.info('All links have been set up')
else:
self._log.error('Some links were not successfully set up')
return success
def _is_link(self, path):
'''
Returns true if the path is a symbolic link.
'''
return os.path.islink(os.path.expanduser(path))
def _link_destination(self, path):
'''
Returns the absolute path to the destination of the symbolic link.
'''
path = os.path.expanduser(path)
rel_dest = os.readlink(path)
return os.path.join(os.path.dirname(path), rel_dest)
def _exists(self, path):
'''
Returns true if the path exists.
'''
path = os.path.expanduser(path)
return os.path.exists(path)
def _link(self, source, link_name):
'''
Links link_name to source.
Returns true if successfully linked files.
'''
success = False
source = os.path.join(self._base_directory, source)
if not self._exists(link_name) and self._is_link(link_name):
self._log.warning('Invalid link %s -> %s' %
(link_name, self._link_destination(link_name)))
elif not self._exists(link_name):
self._log.lowinfo('Creating link %s -> %s' % (link_name, source))
os.symlink(source, os.path.expanduser(link_name))
success = True
elif self._exists(link_name) and not self._is_link(link_name):
self._log.warning(
'%s already exists but is a regular file or directory' %
link_name)
elif self._link_destination(link_name) != source:
self._log.warning('Incorrect link %s -> %s' %
(link_name, self._link_destination(link_name)))
else:
self._log.lowinfo('Link exists %s -> %s' % (link_name, source))
success = True
return success

View file

@ -0,0 +1,2 @@
from .messenger import Messenger
from .level import Level

View file

@ -0,0 +1,8 @@
class Color(object):
NONE = ''
RESET = '\033[0m'
RED = '\033[91m'
GREEN = '\033[92m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
MAGENTA = '\033[95m'

View file

@ -0,0 +1,7 @@
class Level(object):
NOTSET = 0
DEBUG = 10
LOWINFO = 15
INFO = 20
WARNING = 30
ERROR = 40

View file

@ -0,0 +1,60 @@
import sys
from ..util.singleton import Singleton
from .color import Color
from .level import Level
class Messenger(object):
__metaclass__ = Singleton
def __init__(self, level = Level.LOWINFO):
self.set_level(level)
def set_level(self, level):
self._level = level
def log(self, level, message):
if (level >= self._level):
print '%s%s%s' % (self._color(level), message, self._reset())
def debug(self, message):
self.log(Level.DEBUG, message)
def lowinfo(self, message):
self.log(Level.LOWINFO, message)
def info(self, message):
self.log(Level.INFO, message)
def warning(self, message):
self.log(Level.WARNING, message)
def error(self, message):
self.log(Level.ERROR, message)
def _color(self, level):
'''
Get a color (terminal escape sequence) according to a level.
'''
if not sys.stdout.isatty():
return ''
elif level < Level.DEBUG:
return ''
elif Level.DEBUG <= level < Level.LOWINFO:
return Color.YELLOW
elif Level.LOWINFO <= level < Level.INFO:
return Color.BLUE
elif Level.INFO <= level < Level.WARNING:
return Color.GREEN
elif Level.WARNING <= level < Level.ERROR:
return Color.MAGENTA
elif Level.ERROR <= level:
return Color.RED
def _reset(self):
'''
Get a reset color (terminal escape sequence).
'''
if not sys.stdout.isatty():
return ''
else:
return Color.RESET

0
dotbot/util/__init__.py Normal file
View file

6
dotbot/util/singleton.py Normal file
View file

@ -0,0 +1,6 @@
class Singleton(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs)
return cls._instances[cls]