1 """The `command` module provides :class:`Command`\s and
2 :class:`Argument`\s for defining commands.
11 class CommandExit (Exception):
14 class Success (CommandExit):
17 class Failure (CommandExit):
20 class UncaughtException (Failure):
21 def __init__(self, exception):
22 self.exception = exception
23 self.exc_string = traceback.format_exc()
25 super(UncaughtException, self).__init__(self.exc_string)
27 class Command (object):
28 """One-line command description here.
30 >>> c = Command(name='test', help='An example Command.')
31 >>> status = c.run(NullQueue(), PrintQueue(), help=True) # doctest: +REPORT_UDIFF
37 help BOOL (bool) Print a help message.
43 def __init__(self, name, aliases=None, arguments=[], help=''):
47 self.aliases = aliases
49 Argument(name='help', type='bool', default=False, count=1,
50 callback=StoreValue(True), help='Print a help message.'),
54 def run(self, hooke, inqueue=None, outqueue=None, **kwargs):
55 """`Normalize inputs and handle <Argument help> before punting
61 outqueue = NullQueue()
63 params = self.handle_arguments(hooke, inqueue, outqueue, kwargs)
64 if params['help'] == True:
65 outqueue.put(self.help())
67 self._run(hooke, inqueue, outqueue, params)
68 except CommandExit, e:
69 if isinstance(e, Failure):
72 # other CommandExit subclasses fall through to the end
74 x = UncaughtException(e)
82 def _run(self, inqueue, outqueue, params):
83 """This is where the command-specific magic will happen.
87 def handle_arguments(self, hooke, inqueue, outqueue, params):
88 """Normalize and validate input parameters (:class:`Argument` values).
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)
95 if argument.optional == True or argument.count == 0:
96 settings = [(argument.name, argument.default)]
98 raise Failure('Required argument %s not set.'
101 raise Failure('Multiple settings for %s:\n %s'
103 '\n '.join(['%s: %s' % (name,value)
104 for name,value in sorted(settings)])))
105 name,value = settings[0]
106 if name != argument.name:
108 params[argument.name] = value
109 if argument.callback != None:
111 value = argument.callback(hooke, self, argument, value)
112 params[argument.name] = value
113 argument.validate(value)
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)
121 if len(self.arguments) > 0:
122 argument_part = ['Arguments:', '']
123 for a in self.arguments:
124 argument_part.append(textwrap.fill(
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)
133 class Argument (object):
134 """Structured user input for :class:`Command`\s.
136 TODO: ranges for `count`?
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=''):
144 self.aliases = aliases
147 metavar = type.upper()
148 self.metavar = metavar
149 self.default = default
150 self.optional = optional
152 self.completion_callback = completion_callback
153 self.callback = callback
157 return '<%s %s>' % (self.__class__.__name__, self.name)
160 return self.__str__()
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)
169 def validate(self, value):
170 """If `value` is not appropriate, raise `ValueError`.
172 pass # TODO: validation
174 # TODO: type conversion
176 # TODO: type extensions?
180 class StoreValue (object):
181 def __init__(self, value):
183 def __call__(self, hooke, command, argument, fragment=None):
186 class NullQueue (queue.Queue):
187 """The :class:`queue.Queue` equivalent of `/dev/null`.
189 This is a bottomless pit. Items go in, but never come out.
191 def get(self, block=True, timeout=None):
192 """Raise queue.Empty.
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
201 def put(self, item, block=True, timeout=None):
202 """Dump an item into the void.
204 Block and timeout are meaningless, because there is always a
205 free slot available in a bottomless pit.
209 class PrintQueue (NullQueue):
210 """Debugging :class:`NullQueue` that prints items before dropping
213 def put(self, item, block=True, timeout=None):
214 """Print `item` and then dump it into the void.
216 print 'ITEM:\n%s' % item