- params = self.handle_arguments(inqueue, outqueue, kwargs)
- if params['help'] == True:
- outqueue.put(self.help())
- raise(Success())
- self._run(inqueue, outqueue, params)
- except CommandExit, e:
- if isinstance(e, Failure):
- outqueue.put(e.message)
- outqueue.put(e)
- return 1
- outqueue.put(e)
- return 0
-
- def _run(self, inqueue, outqueue, params):
- """This is where the command-specific magic will happen.
- """
- pass
-
- def handle_arguments(self, inqueue, outqueue, params):
- """Normalize and validate input parameters (:class:`Argument` values).
- """
- for argument in self.arguments:
- names = [argument.name] + argument.aliases
- settings = [(name,v) for name,v in params.items() if name in names]
- if len(settings) == 0:
- if argument.optional == True or argument.count == 0:
- settings = [(argument.name, argument.default)]
- else:
- raise Failure('Required argument %s not set.'
- % argument.name)
- if len(settings) > 1:
- raise Failure('Multiple settings for %s:\n %s'
- % (argument.name,
- '\n '.join(['%s: %s' % (name,value)
- for name,value in sorted(settings)])))
- name,value = settings[0]
- if name != argument.name:
- params.remove(name)
- params[argument.name] = value
- if argument.callback != None:
- value = argument.callback(self, argument, value)
- params[argument.name] = value
- argument.validate(value)
- return params
-
- def help(self, *args):
- name_part = 'Command: %s' % self.name
- if len(self.aliases) > 0:
- name_part += ' (%s)' % ', '.join(self.aliases)
- argument_part = ['Arguments:'] + [a.help() for a in self.arguments]
- argument_part = '\n'.join(argument_part)
- help_part = self._help
- return '\n\n'.join([name_part, argument_part, help_part])
-
-class Argument (object):
- """Structured user input for :class:`Command`\s.
-
- TODO: ranges for `count`?
- """
- def __init__(self, name, aliases=None, type='string', metavar=None,
- default=None, optional=True, count=1,
- completion_callback=None, callback=None, help=''):
- self.name = name
- if aliases == None:
- aliases = []
- self.aliases = aliases
- self.type = type
- if metavar == None:
- metavar = type.upper()
- self.metavar = metavar
- self.default = default
- self.optional = optional
- self.count = count
- self.completion_callback = completion_callback
- self.callback = callback
- self._help = help
-
- def __str__(self):
- return '<%s %s>' % (self.__class__.__name__, self.name)
-
- def __repr__(self):
- return self.__str__()
-
- def help(self):
- parts = ['%s ' % self.name]
- if self.metavar != None:
- parts.append('%s ' % self.metavar)
- parts.extend(['(%s) ' % self.type, self._help])
- return ''.join(parts)
-
- def validate(self, value):
- """If `value` is not appropriate, raise `ValueError`.
- """
- pass # TODO: validation
-
- # TODO: type conversion
-
-# TODO: type extensions?
-
-# Useful callbacks
-
-class StoreValue (object):
- def __init__(self, value):
- self.value = value
- def __call__(self, command, argument, fragment=None):
- return self.value
-
-class NullQueue (queue.Queue):
- """The :class:`queue.Queue` equivalent of `/dev/null`.
-
- This is a bottomless pit. Items go in, but never come out.
- """
- def get(self, block=True, timeout=None):
- """Raise queue.Empty.
-
- There's really no need to override the base Queue.get, but I
- want to know if someone tries to read from a NullQueue. With
- the default implementation they would just block silently
- forever :(.
- """
- raise queue.Empty
-
- def put(self, item, block=True, timeout=None):
- """Dump an item into the void.
-
- Block and timeout are meaningless, because there is always a
- free slot available in a bottomless pit.
- """
- pass
-
-class PrintQueue (NullQueue):
- """Debugging :class:`NullQueue` that prints items before dropping
- them.
- """
- def put(self, item, block=True, timeout=None):
- """Print `item` and then dump it into the void.
- """
- print 'ITEM:\n%s' % item