forked from piranha/opster
-
Notifications
You must be signed in to change notification settings - Fork 0
/
opster.py
1096 lines (859 loc) · 33.5 KB
/
opster.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# (c) Alexander Solovyov, 2009-2011, under terms of the new BSD License
'''Command line arguments parser
'''
import sys, traceback, getopt, types, textwrap, inspect, os, keyword, codecs
from itertools import imap
from functools import wraps
from collections import namedtuple, Callable
from contextlib import contextmanager
__all__ = ['Dispatcher', 'command', 'dispatch']
__version__ = '3.7'
__author__ = 'Alexander Solovyov'
__email__ = '[email protected]'
def write(text, out=None):
'''Write output to a given stream (stdout by default).'''
out = out or sys.stdout
try:
print >> out, text
# Needed on Python 2.x if text is str/bytes containing non-ascii
# characters and sys.stdout is replaced by a writer from the codecs
# module. text will be decoded as ascii giving the decode error.
except UnicodeDecodeError:
print >> out, text.decode('utf-8')
# Get the order of stdout/stderr correct on Windows. AFAICT this is only
# needed for the test environment but it's harmless otherwise.
out.flush()
def err(text):
'''Write output to stderr.'''
write(text, out=sys.stderr)
# encoding to use when decoding command line arguments
FSE_ENCODING = sys.getfilesystemencoding()
ARG_ENCODING = os.environ.get('OPSTER_ARG_ENCODING', FSE_ENCODING)
def decodearg(arg, arg_encoding=ARG_ENCODING):
'''Decode an argument from sys.argv'''
# python 2.x: have bytes, convert to unicode with given encoding
if sys.version_info < (3, 0):
return arg.decode(arg_encoding)
# python 3.x: have unicode
# arg has already been decoded with FSE_ENCODING
# In the default case we just return the arg as it is
if arg_encoding == FSE_ENCODING:
return arg
# Need to encode and redecode as arg_encoding
if os.name == 'posix':
# On posix the argument was decoded using surrogate escape
arg = arg.encode(FSE_ENCODING, 'surrogateescape')
else:
# On windows the 'mbcs' codec has no surrogate escape handler
arg = arg.encode(FSE_ENCODING)
return arg.decode(arg_encoding)
class Dispatcher(object):
'''Central object for command dispatching system.
- ``cmdtable``: dict of commands. Will be populated with functions,
decorated with ``Dispatcher.command``.
- ``globaloptions``: list of options which are applied to all
commands, will contain ``--help`` option at least.
- ``middleware``: global decorator for all commands.
'''
def __init__(self, cmdtable=None, globaloptions=None, middleware=None):
self._cmdtable = CmdTable(cmdtable or {})
self._globaloptions = [Option(o) for o in (globaloptions or [])]
self.middleware = middleware
@property
def globaloptions(self):
opts = self._globaloptions[:]
if not any(o.name == 'help' for o in opts):
opts.append(Option(('h', 'help', False, 'display help')))
return opts
@property
def cmdtable(self):
return self._cmdtable.copy()
def command(self, options=None, usage=None, name=None, shortlist=False,
hide=False, aliases=()):
'''Decorator to mark function to be used as command for CLI.
Usage::
from opster import command, dispatch
@command()
def run(argument,
optionalargument=None,
option=('o', 'default', 'help for option'),
no_short_name=('', False, 'help for this option')):
print argument, optionalargument, option, no_short_name
if __name__ == '__main__':
run.command()
# or, if you want to have multiple subcommands:
if __name__ == '__main__':
dispatch()
Optional arguments:
- ``options``: options in format described later. If not supplied,
will be determined from function.
- ``usage``: usage string for function, replaces ``%name`` with name
of program or subcommand. In case if it's subcommand and ``%name``
is not present, usage is prepended by ``name``
- ``name``: used for multiple subcommands. Defaults to wrapped
function name
- ``shortlist``: if command should be included in shortlist. Used
only with multiple subcommands
- ``hide``: if command should be hidden from help listing. Used only
with multiple subcommands, overrides ``shortlist``
- ``aliases``: list of aliases for command
If defined, options should be a list of 4-tuples in format::
(shortname, longname, default, help)
Where:
- ``shortname`` is a single letter which can be used then as an option
specifier on command line (like ``-a``). Will be not used if contains
falsy value (empty string, for example)
- ``longname`` - main identificator of an option, can be used as on a
command line with double dashes (like ``--longname``)
- ``default`` value for an option, type of it determines how option
will be processed
- ``help`` string displayed as a help for an option when asked to
'''
def wrapper(func):
try:
options_ = [Option(o) for o in (options or guess_options(func))]
except TypeError:
options_ = []
cmdname = name or name_from_python(func.__name__)
scriptname_ = name or sysname()
if usage is None:
usage_ = guess_usage(func, options_)
else:
usage_ = usage
prefix = hide and '~' or (shortlist and '^' or '')
cmdname = prefix + cmdname
if aliases:
cmdname = cmdname + '|' + '|'.join(aliases)
self._cmdtable[cmdname] = (func, options_, usage_)
def help_func(scriptname=None):
scriptname = scriptname or sysname()
return help_cmd(func, usage_, options_, aliases, scriptname)
def command(argv=None, scriptname=None):
scriptname = scriptname or sysname()
merge_globalopts(self.globaloptions, options_)
if argv is None:
argv = sys.argv[1:]
try:
with exchandle(func.help, scriptname):
args, opts = process(argv, options_)
if opts.pop('help', False):
return func.help(scriptname)
with exchandle(func.help, scriptname):
with help_workaround(func, scriptname):
return call_cmd(scriptname, func, options_)(*args, **opts)
except ErrorHandled:
return -1
func.usage = usage_
func.help = help_func
func.command = command
func.opts = options_
func.orig = func
func.scriptname = scriptname_
@wraps(func)
def inner(*args, **opts):
return call_cmd_regular(func, options_)(*args, **opts)
# Store this for help_workaround
func._inner = inner
return inner
return wrapper
def nest(self, name, dispatcher, help, hide=False, shortlist=False):
'''Add another dispatcher as a subcommand.'''
dispatcher.__doc__ = help
prefix = hide and '~' or (shortlist and '^' or '')
self._cmdtable[prefix + name] = dispatcher, [], None
def dispatch(self, args=None, scriptname=None):
'''Dispatch command line arguments using subcommands.
- ``args``: list of arguments, default: ``sys.argv[1:]``
'''
if args is None:
args = sys.argv[1:]
scriptname = scriptname or sysname()
# Add help function to the table
cmdtable = self.cmdtable
help_func = help_(cmdtable, self.globaloptions, scriptname)
cmdtable['help'] = help_func, [], '[TOPIC]'
autocomplete(cmdtable, args, self.middleware)
try:
with exchandle(help_func):
cmd, func, args, options = cmdparse(args, cmdtable,
self.globaloptions)
if isinstance(func, Dispatcher):
return func.dispatch(args, scriptname=scriptname + ' ' + cmd)
with exchandle(help_func, cmd):
args, opts = process(args, options)
if not cmd:
cmd, func, args, opts = ('help', help_func, ['shortlist'], {})
if opts.pop('help', False):
cmd, func, args, opts = ('help', help_func, [cmd], {})
mw = cmd != '_completion' and self.middleware or None
with exchandle(help_func, cmd):
with help_workaround(func, cmd, help_func):
return call_cmd(cmd, func, options, mw)(*args, **opts)
except ErrorHandled:
return -1
_dispatcher = None
def command(options=None, usage=None, name=None, shortlist=False, hide=False,
aliases=()):
global _dispatcher
if not _dispatcher:
_dispatcher = Dispatcher()
return _dispatcher.command(options=options, usage=usage, name=name,
shortlist=shortlist, hide=hide, aliases=aliases)
command.__doc__ = Dispatcher.command.__doc__
def dispatch(args=None, cmdtable=None, globaloptions=None, middleware=None,
scriptname=None):
global _dispatcher
if not _dispatcher:
_dispatcher = Dispatcher(cmdtable, globaloptions, middleware)
else:
if cmdtable:
_dispatcher._cmdtable = CmdTable(cmdtable)
if globaloptions:
_dispatcher._globaloptions = [Option(o) for o in globaloptions]
if middleware:
_dispatcher.middleware = middleware
return _dispatcher.dispatch(args, scriptname)
dispatch.__doc__ = Dispatcher.dispatch.__doc__
# --------
# Help
# --------
def help_(cmdtable, globalopts, scriptname):
'''Help generator for a command table.
'''
def help_inner(name=None, *args, **opts):
'''Show help for a given help topic or a help overview.
With no arguments, print a list of commands with short help messages.
Given a command name, print help for that command.
'''
def helplist():
hlp = {}
# determine if any command is marked for shortlist
shortlist = (name == 'shortlist' and
any(imap(lambda x: x.startswith('^'), cmdtable)))
for cmd, info in cmdtable.items():
if cmd.startswith('~'):
continue # do not display hidden commands
if shortlist and not cmd.startswith('^'):
continue # short help contains only marked commands
cmd = cmd.lstrip('^~')
doc = pretty_doc_string(info[0])
hlp[cmd] = doc.strip().splitlines()[0].rstrip()
hlplist = sorted(hlp)
maxlen = max(map(len, hlplist))
write('usage: %s <command> [options]' % scriptname)
write('\ncommands:\n')
for cmd in hlplist:
doc = hlp[cmd]
write(' %-*s %s' % (maxlen, cmd.split('|', 1)[0], doc))
if not cmdtable:
return err('No commands specified!')
if not name or name == 'shortlist':
return helplist()
aliases, (cmd, options, usage) = findcmd(name, cmdtable)
if isinstance(cmd, Dispatcher):
recurse = help_(cmd.cmdtable, globalopts, scriptname + ' ' + name)
return recurse(*args, **opts)
options = list(options)
merge_globalopts(globalopts, options)
return help_cmd(cmd, usage, options, aliases[1:],
scriptname + ' ' + aliases[0])
return help_inner
def help_cmd(func, usage, options, aliases, scriptname=None):
'''Show help for given command.
- ``func``: function to generate help for (``func.__doc__`` is taken)
- ``usage``: usage string
- ``options``: options in usual format
>>> def test(*args, **opts):
... """that's a test command
...
... you can do nothing with this command"""
... pass
>>> opts = [('l', 'listen', 'localhost',
... 'ip to listen on'),
... ('p', 'port', 8000,
... 'port to listen on'),
... ('d', 'daemonize', False,
... 'daemonize process'),
... ('', 'pid-file', '',
... 'name of file to write process ID to')]
>>> help_cmd(test, '%name [-l HOST] [NAME]', opts, (), 'test')
test [-l HOST] [NAME]
<BLANKLINE>
that's a test command
<BLANKLINE>
you can do nothing with this command
<BLANKLINE>
options:
<BLANKLINE>
-l --listen ip to listen on (default: localhost)
-p --port port to listen on (default: 8000)
-d --daemonize daemonize process
--pid-file name of file to write process ID to
'''
options = [Option(o) for o in options] # only for doctest
usage = replace_name(usage, scriptname)
write(usage)
if aliases:
write('\naliases: ' + ', '.join(aliases))
doc = pretty_doc_string(func)
write('\n' + doc.strip() + '\n')
for line in help_options(options):
write(line)
def help_options(options):
'''Generator for help on options.
'''
yield 'options:\n'
output = []
for o in options:
default = o.default_value()
default = default and ' (default: %s)' % default or ''
output.append(('%2s%s' % (o.short and '-%s' % o.short,
o.name and ' --%s' % o.name),
'%s%s' % (o.helpmsg, default)))
opts_len = max([len(first) for first, second in output if second] or [0])
for first, second in output:
if second:
# wrap description at 78 chars
second = textwrap.wrap(second, width=(78 - opts_len - 3))
pad = '\n' + ' ' * (opts_len + 3)
yield ' %-*s %s' % (opts_len, first, pad.join(second))
else:
yield ' %s' % first
# --------
# Options process
# --------
def merge_globalopts(globalopts, opts):
'''Merge the global options with the subcommand options'''
for o in globalopts:
# Don't include global option if long name matches
if any((x.name == o.name for x in opts)):
continue
# Don't use global option short name if already used
if any((x.short and x.short == o.short for x in opts)):
o = o._replace(short='')
opts.append(o)
# Factory for creating _Option instances. Intended to be the entry point to
# the *Option classes here.
def Option(opt):
'''Create Option instance from tuple of option data.'''
if isinstance(opt, BaseOption):
return opt
# Extract and validate contents of tuple
short, name, default, helpmsg = opt[:4]
completer = opt[4] if len(opt) > 4 else None
if short and len(short) != 1:
raise OpsterError(
'Short option should be only a single character: %s' % short)
if not name:
raise OpsterError(
'Long name should be defined for every option')
pyname = name_to_python(name)
args = pyname, name, short, default, helpmsg, completer
# Find matching _Option subclass and return instance
# nb. the order of testing matters
for Type in (BoolOption, ListOption, DictOption, FuncOption,
TupleOption, UnicodeOption, LiteralOption):
if Type.matches(default):
return Type(*args)
raise OpsterError('Cannot figure out type for option %s' % name)
def CmdTable(cmdtable):
'''Factory to convert option tuples in a cmdtable'''
newtable = {}
for name, (func, opts, usage) in cmdtable.items():
newtable[name] = (func, [Option(o) for o in opts], usage)
return newtable
# Superclass for all option classes
class BaseOption(namedtuple('Option', (
'pyname', 'name', 'short', 'default', 'helpmsg', 'completer'))):
has_parameter = True
type = None
def __repr__(self):
return (super(BaseOption, self).__repr__()
.replace('Option', self.__class__.__name__, 1))
@classmethod
def matches(cls, default):
'''Returns True if this is appropriate Option for the default value.'''
return isinstance(default, cls.type)
def default_state(self):
'''Generate initial state value from provided default value.'''
return self.default
def update_state(self, state, new):
'''Update state after encountering an option on the command line.'''
return new
def convert(self, final):
'''Generate the resulting python value from the final state.'''
return final
def default_value(self):
'''Shortcut to obtain the default value when option arg not provided.'''
return self.convert(self.default_state())
class LiteralOption(BaseOption):
'''Literal option type (including string, int, float, etc.)'''
type = object
def convert(self, final):
'''Generate the resulting python value from the final state.'''
if final is self.default:
return final
else:
return type(self.default)(final)
class UnicodeOption(BaseOption):
'''Handle unicode values, decoding input'''
type = unicode
def convert(self, final):
return decodearg(final)
class BoolOption(BaseOption):
'''Boolean option type.'''
has_parameter = False
type = (bool, types.NoneType)
def convert(self, final):
return bool(final)
def update_state(self, state, new):
return not self.default
class ListOption(BaseOption):
'''List option type.'''
type = list
def default_state(self):
return list(self.default)
def update_state(self, state, new):
state.append(new)
return state
class DictOption(BaseOption):
'''Dict option type.'''
type = dict
def default_state(self):
return dict(self.default)
def update_state(self, state, new):
try:
k, v = new.split('=')
except ValueError:
msg = "wrong definition: %r (should be in format KEY=VALUE)"
raise getopt.GetoptError(msg % new)
state[k] = v
return state
class TupleOption(BaseOption):
'''Tuple option type.'''
type = tuple
def __init__(self, *args, **kwargs):
self._option = Option(('', '_', self.default[0], ''))
def default_state(self):
return self._option.default
def update_state(self, state, new):
return self._option.update_state(state, new)
def convert(self, final):
finalval = self._option.convert(final)
if finalval not in self.default:
msg = "unrecognised value: %r (should be one of %s)"
msg = msg % (final, ', '.join(str(v) for v in self.default))
raise getopt.GetoptError(msg)
return finalval
class FuncOption(BaseOption):
'''Function option type.'''
type = Callable
def default_state(self):
return None
def convert(self, final):
return self.default(final)
def process(args, options):
'''
>>> opts = [('l', 'listen', 'localhost',
... 'ip to listen on'),
... ('p', 'port', 8000,
... 'port to listen on'),
... ('d', 'daemonize', False,
... 'daemonize process'),
... ('', 'pid-file', '',
... 'name of file to write process ID to')]
>>> print process(['-l', '0.0.0.0', '--pi', 'test', 'all'], opts)
(['all'], {'pid_file': 'test', 'daemonize': False, 'port': 8000, 'listen': '0.0.0.0'})
'''
options = [Option(o) for o in options] # only for doctest
# Parse arguments and options
args, opts = getopts(args, options)
# Default values
state = dict((o.pyname, o.default_state()) for o in options)
# Update for each option on the command line
for o, val in opts:
state[o.pyname] = o.update_state(state[o.pyname], val)
# Convert to required type
for o in options:
try:
state[o.pyname] = o.convert(state[o.pyname])
except ValueError:
raise getopt.GetoptError('invalid option value %r for option %r'
% (state[o.pyname], o.name))
return args, state
def getopts(args, options, preparse=False):
'''Parse args and options from raw args.
If preparse is True, option processing stops at first non-option.
'''
argmap = {}
shortlist, namelist = '', []
for o in options:
argmap['-' + o.short] = argmap['--' + o.name] = o
# getopt wants indication that it takes a parameter
short, name = o.short, o.name
if o.has_parameter:
if short:
short += ':'
name += '='
if short:
shortlist += short
namelist.append(name)
# gnu_getopt will stop at first non-option argument
if preparse:
shortlist = '+' + shortlist
# getopt.gnu_getopt allows options after the first non-option
opts, args = getopt.gnu_getopt(args, shortlist, namelist)
# map the option argument names back to their Option instances
opts = [(argmap[opt], val) for opt, val in opts]
return args, opts
# --------
# Subcommand system
# --------
def cmdparse(args, cmdtable, globalopts):
'''Parse arguments list to find a command, options and arguments.
'''
# pre-parse arguments here using global options to find command name,
# which is first non-option entry
args_new, opts = getopts(args, globalopts, preparse=True)
args = list(args)
if args_new:
cmdarg = args_new[0]
args.remove(cmdarg)
aliases, info = findcmd(cmdarg, cmdtable)
cmd = aliases[0]
possibleopts = list(info[1])
merge_globalopts(globalopts, possibleopts)
return cmd, info[0] or None, args, possibleopts
else:
return None, None, args, globalopts
def aliases_(cmdtable_key):
'''Get aliases from a command table key.'''
return cmdtable_key.lstrip("^~").split("|")
def findpossible(cmd, table):
'''Return cmd -> (aliases, command table entry) for each matching command.
'''
choice = {}
for e in table.keys():
aliases = aliases_(e)
found = None
if cmd in aliases:
found = cmd
else:
for a in aliases:
if a.startswith(cmd):
found = a
break
if found is not None:
choice[found] = (aliases, table[e])
return choice
def findcmd(cmd, table):
"""Return (aliases, command table entry) for command string."""
choice = findpossible(cmd, table)
if cmd in choice:
return choice[cmd]
if len(choice) > 1:
clist = choice.keys()
clist.sort()
raise AmbiguousCommand(cmd, clist)
if choice:
return choice.values()[0]
raise UnknownCommand(cmd)
# --------
# Helpers
# --------
def guess_options(func):
'''Get options definitions from function
They should be declared in a following way:
def func(longname=(shortname, default, help)):
pass
See docstring of ``command()`` for description of those variables.
'''
try:
args, _, _, defaults = inspect.getargspec(func)
except ValueError: # has keyword-only arguments
for o in guess_options_py3k(func):
yield o
return
for name, option in zip(args[-len(defaults):], defaults):
if not isinstance(option, tuple):
continue
yield (option[0], name_from_python(name)) + option[1:]
def guess_options_py3k(func):
'''Get options definitions from function with keyword-only arguments
They should be declared in a following way:
def func(*, longname=(shortname, default, help)):
pass
See docstring of ``command()`` for description of those variables.
'''
spec = inspect.getfullargspec(func)
for name, option in spec.kwonlydefaults.items():
yield (option[0], name_from_python(name)) + option[1:]
def guess_usage(func, options):
'''Get usage definition for a function
'''
usage = ['%name']
if options:
usage.append('[OPTIONS]')
try:
arginfo = inspect.getargspec(func)
except ValueError: # keyword-only args
arginfo = inspect.getfullargspec(func)
optnames = [o.name for o in options]
nonoptional = len(arginfo.args) - len(arginfo.defaults or ())
for i, arg in enumerate(arginfo.args):
if arg not in optnames:
usage.append((i > nonoptional - 1 and '[%s]' or '%s')
% arg.upper())
if arginfo.varargs:
usage.append('[%s ...]' % arginfo.varargs.upper())
return ' '.join(usage)
@contextmanager
def help_workaround(func, scriptname, help_func=None):
'''Context manager to temporarily replace func.help'''
# Retrieve inner if function is command wrapped
func = getattr(func, '_inner', func)
# Ignore function that was not command wrapped
if not hasattr(func, 'help'):
yield
return
# Wrap the with block with a replaced help function
help = func.help
help_func = help_func or help
try:
func.help = lambda: help_func(scriptname)
yield
finally:
func.help = help
@contextmanager
def exchandle(help_func, cmd=None):
'''Context manager to turn internal exceptions into printed help messages.
Handles internal opster exceptions by printing help and raising
ErrorHandled. Other exceptions are propagated.
'''
try:
yield # execute the block in the 'with' statement
return
except UnknownCommand as e:
err("unknown command: '%s'" % e)
except AmbiguousCommand as e:
err("command '%s' is ambiguous:\n %s" %
(e.args[0], ' '.join(e.args[1])))
except ParseError as e:
err('%s: %s\n' % (e.args[0], e.args[1].strip()))
help_func(cmd)
except getopt.GetoptError as e:
err('error: %s\n' % e)
help_func(cmd)
except OpsterError as e:
err('%s' % e)
# abort if a handled exception was raised
raise ErrorHandled()
def call_cmd(name, func, opts, middleware=None):
'''Wrapper for command call, catching situation with insufficient arguments.
'''
# depth is necessary when there is a middleware in setup
try:
arginfo = inspect.getargspec(func)
except ValueError:
arginfo = inspect.getfullargspec(func)
if middleware:
tocall = middleware(func)
depth = 2
else:
tocall = func
depth = 1
def inner(*args, **kwargs):
# NOTE: this is not very nice, but it fixes problem with
# TypeError: func() got multiple values for 'argument'
# Would be nice to find better way
prepend = []
start = None
if arginfo.varargs and len(args) > (len(arginfo.args) - len(kwargs)):
for o in opts:
if o.pyname in arginfo.args:
if start is None:
start = arginfo.args.index(o.pyname)
prepend.append(o.pyname)
if start is not None: # do we have to prepend anything
args = (args[:start] +
tuple(kwargs.pop(x) for x in prepend) +
args[start:])
try:
return tocall(*args, **kwargs)
except TypeError:
if len(traceback.extract_tb(sys.exc_info()[2])) == depth:
raise ParseError(name, "invalid arguments")
raise
return inner
def call_cmd_regular(func, opts):
'''Wrapper for command for handling function calls from Python.
'''
# This would raise an error if there were any keyword-only arguments
try:
arginfo = inspect.getargspec(func)
except ValueError:
return call_cmd_regular_py3k(func, opts)
def inner(*args, **kwargs):
# Map from argument names to Option instances
opt_args = dict((o.pyname, o) for o in opts)
# Pull any recognised args out of kwargs and splice them with the
# positional arguments to give a flat positional arg list
remaining = list(args)
args = []
defaults_offset = len(arginfo.args) - len(arginfo.defaults)
for n, argname in enumerate(arginfo.args):
# Option arguments MUST be given as keyword arguments
if argname in opt_args:
if argname in kwargs:
argval = kwargs.pop(argname)
else:
argval = opt_args[argname].default_value()
# Take a positional argument
elif remaining:
argval = remaining.pop(0)
# Find the default value of the positional argument
elif n >= defaults_offset:
argval = arginfo.defaults[n - defaults_offset]
else:
raise TypeError('Not enough positional arguments')
# Accumulate the args in order
args.append(argval)
# Combine the remaining positional arguments that go to varargs
args = args + remaining
# kwargs is any keyword arguments that were not recognised as options
return func(*args, **kwargs)
return inner
def call_cmd_regular_py3k(func, opts):
'''call_cmd_regular for functions with keyword only arguments'''
spec = inspect.getfullargspec(func)
def inner(*args, **kwargs):
# Replace the option arguments with their default values
for o in opts:
if o.pyname not in kwargs:
kwargs[o.pyname] = o.default_value()
return func(*args, **kwargs)
return inner
def replace_name(usage, name):
'''Replace name placeholder with a command name.'''
if '%name' in usage:
return usage.replace('%name', name, 1)
return name + ' ' + usage
def sysname():
'''Returns name of executing file.'''
name = sys.argv[0]
if os.path.isabs(name) or os.path.dirname(name) == '.':
return os.path.basename(name)
return name
def pretty_doc_string(item):
'''Doc string with adjusted indentation level of the 2nd line and beyond.'''
raw_doc = item.__doc__ or '(no help text available)'
lines = raw_doc.strip().splitlines()
if len(lines) <= 1:
return raw_doc
indent = len(lines[1]) - len(lines[1].lstrip())
return '\n'.join([lines[0]] + map(lambda l: l[indent:], lines[1:]))
def name_from_python(name):
if name.endswith('_') and keyword.iskeyword(name[:-1]):
name = name[:-1]
return name.replace('_', '-')
def name_to_python(name):
name = name.replace('-', '_')
if keyword.iskeyword(name):
return name + '_'
return name
# --------
# Autocomplete system
# --------
# Borrowed from PIP
def autocomplete(cmdtable, args, middleware):
'''Command and option completion.
Enable by sourcing one of the completion shell scripts (bash or zsh).
'''
# Don't complete if user hasn't sourced bash_completion file.
if 'OPSTER_AUTO_COMPLETE' not in os.environ:
return
cwords = os.environ['COMP_WORDS'].split()[1:]
cword = int(os.environ['COMP_CWORD'])
try:
current = cwords[cword - 1]
except IndexError:
current = ''
commands = []
for k in cmdtable.keys():
commands += aliases_(k)
# command
if cword == 1:
print ' '.join(filter(lambda x: x.startswith(current), commands))
# command options
elif cwords[0] in commands:
idx = -2 if current else -1