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