-
Notifications
You must be signed in to change notification settings - Fork 1
/
__init__.py
123 lines (94 loc) · 3.35 KB
/
__init__.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
# -*- coding: utf-8 -*-
from __future__ import absolute_import
import inspect
import sys
import types
import warnings
from fabric.tasks import WrappedCallableTask
def task_method(*args, **kwargs):
"""
Decorator declaring the wrapped method to be task.
It acceps the same arguments as ``fabric.decorators.task`` so
use it on methods just like fabric's decorator is used on functions.
The class decorated method belongs to should be a subclass
of :class:`.TaskSet`.
"""
invoked = bool(not args or kwargs)
if not invoked:
func, args = args[0], ()
def decorator(func):
func._task_info = dict(
args = args,
kwargs = kwargs
)
return func
return decorator if invoked else decorator(func)
def task(*args, **kwargs):
msg = "@taskset.task decorator is deprecated and will be removed soon; please use @taskset.task_method instead."
warnings.warn(msg, DeprecationWarning)
return task_method(*args, **kwargs)
class TaskSet(object):
"""
TaskSet is a class that can expose its methods as Fabric tasks.
Example::
# my_lib/say.py
from taskset import TaskSet, task
class SayBase(TaskSet):
def say(self, what):
raise NotImplemented()
@task(default=True, alias='hi')
def hello(self):
self.say('hello')
class EchoSay(SayBase):
def say(self, what):
local('echo ' + what)
instance = EchoSay()
instance.expose_to_current_module()
# fabfile.py
from mylib import say
and then::
$ fab say.hi
"""
def expose_to(self, module_name):
"""
Adds tasks to module which name is passed in ``module_name`` argument.
Returns a list of added tasks names.
Example::
instance = MyTaskSet()
__all__ = instance.expose_to(__name__)
"""
return list(self._expose_to(module_name))
def expose_to_current_module(self):
"""
The same as :meth:`TaskSet.expose_to` but magically
addds tasks to current module.
Example::
instance = MyTaskSet()
__all__ = instance.expose_to_current_module()
"""
frm = inspect.stack()[1]
mod = inspect.getmodule(frm[0])
return self.expose_to(mod.__name__)
def expose_as_module(self, module_name, module_type=types.ModuleType):
"""
Creates a new module of type ``module_type`` and named ``module_name``,
populates it with tasks and returns this newly created module.
"""
module = module_type(module_name)
for name, task in self._get_fabric_tasks():
setattr(module, name, task)
return module
def _expose_to(self, module_name):
module_obj = sys.modules[module_name]
for name, task in self._get_fabric_tasks():
setattr(module_obj, name, task)
yield name
def _is_task(self, func):
return hasattr(func, '_task_info')
def _task_for_method(self, method):
return WrappedCallableTask(method, *method._task_info['args'], **method._task_info['kwargs'])
def _get_fabric_tasks(self):
return (
(name, self._task_for_method(task))
for name, task in inspect.getmembers(self, self._is_task)
)