-
Notifications
You must be signed in to change notification settings - Fork 1
/
celery.py
185 lines (152 loc) · 5.73 KB
/
celery.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
"""Module with function decorator to register functions as commandline
commands and to generate a commandline help text.
Adapted from @uniphil's code for Qcumber API
https://github.com/Queens-Hacks/qcumber-api/blob/master/manage.py
Copyright 2014 jameh and other contributors
* Released under the MIT license
* https://github.com/jameh/celery/blob/master/LICENSE
Usage:
.. code-block::
from celery import command, parse_args, help
@command
def my_func(arg_1, arg_2):
\"\"\"does something cool\"\"\"
return int(arg_1) ^ int(arg_2)
if __name__ == '__main__':
parse_args()
"""
from functools import wraps
import sys
import inspect
import __main__
def command(func, _funcs={}):
"""Decorate functions with this to register them as commands"""
# register the command
func_name = func.__name__.lower()
if func_name in _funcs:
raise Exception('Duplicate definition for command {}'.format(func_name))
if func.__module__ not in _funcs:
_funcs[func.__module__] = {}
_funcs[func.__module__][func_name] = func
# play nice and leave the command where it was in this script
@wraps(func)
def wrapped(*args, **kwargs):
return func(*args, **kwargs)
return wrapped
def parse_args():
"""Get the command, or run 'help' if no command is provided
.. note::
The "=" sign is reserved to indicate keyword arguments e.g.
.. code::
python my_file.py my_func kw_1=3 lala
python my_file.py my_func --kw_1 3 lala
python my_file.py my_func -kw_1 3 lala
python my_file.py my_func --kw_1=3 lala
python my_file.py my_func -kw_1=3 lala
Here, "lala" is the first positional argument, kw_1 is a
keyword argument with value "3"
"""
if len(sys.argv) < 2:
cmd, raw_args = 'help', []
else:
cmd, raw_args = sys.argv[1].lower(), sys.argv[2:]
# Parse raw_args into *args and **kwargs
args = []
kwargs = {}
iter_args = iter(raw_args)
for arg in iter_args:
if arg[:2] == "--":
if len(arg.split("=")) == 1:
# eat the next arg for value
kwargs[arg[2:]] = next(iter_args, "")
elif len(arg[2:].split("=")) == 2:
# use the rhs of "=" as value
kwargs[arg[2:].split("=")[0]] = arg[2:].split("=")[1]
else:
raise Exception('Invalid kwarg format "{}"'.format(arg))
elif arg[:1] == "-":
if len(arg.split("=")) == 1:
# eat the next arg for value
kwargs[arg[1:]] = next(iter_args, "")
elif len(arg[1:].split("=")) == 2:
# use the rhs of "=" as value
kwargs[arg[1:].split("=")[0]] = arg[1:].split("=")[1]
else:
raise Exception('Invalid kwarg format "{}"'.format(arg))
elif len(arg.split("=")) == 2:
kwargs[arg.split("=")[0]] = arg.split("=")[1]
else:
# positional arguments
args.append(arg)
# Map the command to a function, falling back to 'help' if it's not found
funcs = command.__defaults__[0]['__main__'] # _funcs={}
if cmd == 'help':
output = command.__defaults__[0][__name__]['help']()
print(output)
return
elif cmd not in funcs:
output = command.__defaults__[0][__name__]['help'](cmd)
print(output)
return
# do it!
try:
output = funcs[cmd](*args, **kwargs)
# (and print the output)
if output:
print(output)
except Exception as e:
help()
raise e
def _signature(name, func):
"""Return string repr of function signature"""
defaults = inspect.getfullargspec(func).defaults or []
args = inspect.getfullargspec(func).args or []
arg_str_list = []
n_positional_args = len(args) - len(defaults)
for i in range(n_positional_args):
# positional arguments
arg_str_list.append('<' + args[i] + '>')
for i in range(len(defaults)):
# keyword arguments
arg_str_list.append('[' + args[i + n_positional_args] + '=' + str(defaults[i]) + ']')
return '{} {}'.format(name, ' '.join(arg_str_list))
def _indent(string, spaces=4, bullet='|'):
lines = string.splitlines()
for i, line in enumerate(lines):
if line[:4] == ' ' * len(line[:4]):
# starts with 4 spaces, indent and add a '|'
lines[i] = ' ' * spaces + bullet + ' ' + line[4:]
else:
lines[i] = ' ' * spaces + bullet + ' ' + line
return '\n'.join(lines)
@command
def help(*args, **kwargs):
"""Get usage information about this script
Multiple lines!"""
text = ""
for i, f_name in enumerate(args):
# Find f_name in available commands
try:
func = command.__defaults__[0]['__main__'][f_name]
text += "Usage: {} {}\n".format(sys.argv[0],
_signature(f_name, func))
if func.__doc__:
text += _indent(func.__doc__.strip(), spaces=2) + '\n'
return text
except KeyError:
text += help() + '\n'
text += 'Command "{}" not found :('.format(f_name)
return text.strip()
text += 'Usage: {} [command]\n'.format(sys.argv[0])
# print module help here
if __main__.__doc__ is not None:
text += _indent(__main__.__doc__.strip(), spaces=2)
text += '\n'
text += 'Available commands:\n'
for name, func in sorted(command.__defaults__[0]['__main__'].items()): # _funcs={}
text += _indent(_signature(name, func), 2, '*') + '\n'
if func.__doc__:
text += _indent(func.__doc__.strip(), 4, '|') + '\n'
return text.strip()
if __name__ == '__main__':
parse_args()