hooke/command.py

   1 """The `command` module provides :class:`Command`\s and
   2 :class:`Argument`\s for defining commands.
   3 """
   4
   5 import Queue as queue
   6 import sys
   7 import textwrap
   8 import traceback
   9
  10
  11 class CommandExit (Exception):
  12     pass
  13
  14 class Success (CommandExit):
  15     pass
  16
  17 class Failure (CommandExit):
  18     pass
  19
  20 class UncaughtException (Failure):
  21     def __init__(self, exception):
  22         self.exception = exception
  23         self.exc_string = traceback.format_exc()
  24         sys.exc_clear()
  25         super(UncaughtException, self).__init__(self.exc_string)
  26
  27 class Command (object):
  28     """One-line command description here.
  29
  30     >>> c = Command(name='test', help='An example Command.')
  31     >>> status = c.run(NullQueue(), PrintQueue(), help=True) # doctest: +REPORT_UDIFF
  32     ITEM:
  33     Command: test
  34     <BLANKLINE>
  35     Arguments:
  36     <BLANKLINE>
  37     help BOOL (bool) Print a help message.
  38     <BLANKLINE>
  39     An example Command.
  40     ITEM:
  41     Success
  42     """
  43     def __init__(self, name, aliases=None, arguments=[], help=''):
  44         self.name = name
  45         if aliases == None:
  46             aliases = []
  47         self.aliases = aliases
  48         self.arguments = [
  49             Argument(name='help', type='bool', default=False, count=1,
  50                      callback=StoreValue(True), help='Print a help message.'),
  51             ] + arguments
  52         self._help = help
  53
  54     def run(self, hooke, inqueue=None, outqueue=None, **kwargs):
  55         """`Normalize inputs and handle <Argument help> before punting
  56         to :meth:`_run`.
  57         """
  58         if inqueue == None:
  59             inqueue = NullQueue()
  60         if outqueue == None:
  61             outqueue = NullQueue()
  62         try:
  63             params = self.handle_arguments(hooke, inqueue, outqueue, kwargs)
  64             if params['help'] == True:
  65                 outqueue.put(self.help())
  66                 raise(Success())
  67             self._run(hooke, inqueue, outqueue, params)
  68         except CommandExit, e:
  69             if isinstance(e, Failure):
  70                 outqueue.put(e)
  71                 return 1
  72             # other CommandExit subclasses fall through to the end
  73         except Exception, e:
  74             x = UncaughtException(e)
  75             outqueue.put(x)
  76             return 1
  77         else:
  78             e = Success()
  79         outqueue.put(e)
  80         return 0
  81
  82     def _run(self, inqueue, outqueue, params):
  83         """This is where the command-specific magic will happen.
  84         """
  85         pass
  86
  87     def handle_arguments(self, hooke, inqueue, outqueue, params):
  88         """Normalize and validate input parameters (:class:`Argument` values).
  89         """
  90         for argument in self.arguments:
  91             names = [argument.name] + argument.aliases
  92             settings = [(name,v) for name,v in params.items() if name in names]
  93             num_provided = len(settings)
  94             if num_provided == 0:
  95                 if argument.optional == True or argument.count == 0:
  96                     settings = [(argument.name, argument.default)]
  97                 else:
  98                     raise Failure('Required argument %s not set.'
  99                                   % argument.name)
 100             if num_provided > 1:
 101                 raise Failure('Multiple settings for %s:\n  %s'
 102                     % (argument.name,
 103                        '\n  '.join(['%s: %s' % (name,value)
 104                                     for name,value in sorted(settings)])))
 105             name,value = settings[0]
 106             if name != argument.name:
 107                 params.remove(name)
 108                 params[argument.name] = value
 109             if argument.callback != None:
 110                 if num_provided > 0:
 111                     value = argument.callback(hooke, self, argument, value)
 112                 params[argument.name] = value
 113             argument.validate(value)
 114         return params
 115
 116     def help(self, *args):
 117         name_part = 'Command: %s' % self.name
 118         if len(self.aliases) > 0:
 119             name_part += ' (%s)' % ', '.join(self.aliases)
 120         parts = [name_part]
 121         if len(self.arguments) > 0:
 122             argument_part = ['Arguments:', '']
 123             for a in self.arguments:
 124                 argument_part.append(textwrap.fill(
 125                         a.help(),
 126                         initial_indent="",
 127                         subsequent_indent="    "))
 128             argument_part = '\n'.join(argument_part)
 129             parts.append(argument_part)
 130         parts.append(self._help) # help part
 131         return '\n\n'.join(parts)
 132
 133 class Argument (object):
 134     """Structured user input for :class:`Command`\s.
 135
 136     TODO: ranges for `count`?
 137     """
 138     def __init__(self, name, aliases=None, type='string', metavar=None,
 139                  default=None, optional=True, count=1,
 140                  completion_callback=None, callback=None, help=''):
 141         self.name = name
 142         if aliases == None:
 143             aliases = []
 144         self.aliases = aliases
 145         self.type = type
 146         if metavar == None:
 147             metavar = type.upper()
 148         self.metavar = metavar
 149         self.default = default
 150         self.optional = optional
 151         self.count = count
 152         self.completion_callback = completion_callback
 153         self.callback = callback
 154         self._help = help
 155
 156     def __str__(self):
 157         return '<%s %s>' % (self.__class__.__name__, self.name)
 158
 159     def __repr__(self):
 160         return self.__str__()
 161
 162     def help(self):
 163         parts = ['%s ' % self.name]
 164         if self.metavar != None:
 165             parts.append('%s ' % self.metavar)
 166         parts.extend(['(%s) ' % self.type, self._help])
 167         return ''.join(parts)
 168
 169     def validate(self, value):
 170         """If `value` is not appropriate, raise `ValueError`.
 171         """
 172         pass # TODO: validation
 173
 174     # TODO: type conversion
 175
 176 # TODO: type extensions?
 177
 178 # Useful callbacks
 179
 180 class StoreValue (object):
 181     def __init__(self, value):
 182         self.value = value
 183     def __call__(self, hooke, command, argument, fragment=None):
 184         return self.value
 185
 186 class NullQueue (queue.Queue):
 187     """The :class:`queue.Queue` equivalent of `/dev/null`.
 188
 189     This is a bottomless pit.  Items go in, but never come out.
 190     """
 191     def get(self, block=True, timeout=None):
 192         """Raise queue.Empty.
 193
 194         There's really no need to override the base Queue.get, but I
 195         want to know if someone tries to read from a NullQueue.  With
 196         the default implementation they would just block silently
 197         forever :(.
 198         """
 199         raise queue.Empty
 200
 201     def put(self, item, block=True, timeout=None):
 202         """Dump an item into the void.
 203
 204         Block and timeout are meaningless, because there is always a
 205         free slot available in a bottomless pit.
 206         """
 207         pass
 208
 209 class PrintQueue (NullQueue):
 210     """Debugging :class:`NullQueue` that prints items before dropping
 211     them.
 212     """
 213     def put(self, item, block=True, timeout=None):
 214         """Print `item` and then dump it into the void.
 215         """
 216         print 'ITEM:\n%s' % item