如何创建 fabric2 任务装饰器?



让我们首先考虑一些使用 fabric2 的未重构工作代码:

import traceback import inspect

from fabric2.tasks import task 
from fabric2 import Config, Connection

def get_function_name():
return traceback.extract_stack(None, 2)[0][2]

def get_function_parameters_and_values():
frame = inspect.currentframe().f_back
args, _, _, values = inspect.getargvalues(frame)
return ", ".join([str(i) for i in args])

@task def foo1(c, arg1):
print(f'<{get_function_name()} {get_function_parameters_and_values()}>')
def _inner():
if int(arg1) < 5:
print('arg1 is not big enough')
return
print('doing foo1 stuff')
res = _inner()
print(f'</{get_function_name}')
return res

@task def foo2(c, arg1, arg2):
print(f'<{get_function_name()} {get_function_parameters_and_values()}>')
def _inner():
if int(arg1) < 5 and int(arg2) < 5:
print('arg1 & arg2 are not big enough')
return
print('doing foo2 stuff')
res = _inner()
print(f'</{get_function_name}')
return res

现在,正如您在上面看到的,我想以包装器/装饰器的方式排除一些代码重复......如果我尝试这样的事情,问题就来了:

def wrap(func):
def wrapped(*args, **kwargs):
print('begin')
res = func(*args, **kwargs)
print('end')
return wrapped

@task
@wrap
def foo3(c, arg1):
if int(arg1) < 5:
print('arg1 is not big enough')
return
print('doing foo3 stuff')

@wrap
@task
def foo4(c, arg1):
if int(arg1) < 5:
print('arg1 is not big enough')
return
print('doing foo4 stuff')

我运行这些任务中的任何一个,我都会得到两个任务的异常,例如:

(py364_32) D:sourcespersonalpythonbplfabexamples>fab2 foo3 --arg1=3
Traceback (most recent call last):
File "d:virtual_envspy364_32libsite-packagesinvoketasks.py", line 160, in argspec
context_arg = arg_names.pop(0)
IndexError: pop from empty list
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "d:softwarepython364_32Librunpy.py", line 193, in _run_module_as_main
"__main__", mod_spec)
File "d:softwarepython364_32Librunpy.py", line 85, in _run_code
exec(code, run_globals)
File "D:virtual_envspy364_32Scriptsfab2.exe__main__.py", line 7, in <module>
File "d:virtual_envspy364_32libsite-packagesinvokeprogram.py", line 352, in run
self.parse_collection()
File "d:virtual_envspy364_32libsite-packagesinvokeprogram.py", line 444, in parse_collection
self.load_collection()
File "d:virtual_envspy364_32libsite-packagesfabric2main.py", line 87, in load_collection
super(Fab, self).load_collection()
File "d:virtual_envspy364_32libsite-packagesinvokeprogram.py", line 661, in load_collection
module, parent = loader.load(coll_name)
File "d:virtual_envspy364_32libsite-packagesinvokeloader.py", line 76, in load
module = imp.load_module(name, fd, path, desc)
File "d:virtual_envspy364_32libimp.py", line 235, in load_module
return load_source(name, filename, file)
File "d:virtual_envspy364_32libimp.py", line 172, in load_source
module = _load(spec)
File "<frozen importlib._bootstrap>", line 684, in _load
File "<frozen importlib._bootstrap>", line 665, in _load_unlocked
File "<frozen importlib._bootstrap_external>", line 678, in exec_module
File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
File "D:sourcespersonalpythonbplfabexamplesfabfile.py", line 59, in <module>
@wrap
File "d:virtual_envspy364_32libsite-packagesfabric2tasks.py", line 71, in task
return invoke.task(*args, **kwargs)
File "d:virtual_envspy364_32libsite-packagesinvoketasks.py", line 313, in task
return klass(args[0], **kwargs)
File "d:virtual_envspy364_32libsite-packagesfabric2tasks.py", line 21, in __init__
super(Task, self).__init__(*args, **kwargs)
File "d:virtual_envspy364_32libsite-packagesinvoketasks.py", line 77, in __init__
self.positional = self.fill_implicit_positionals(positional)
File "d:virtual_envspy364_32libsite-packagesinvoketasks.py", line 168, in fill_implicit_positionals
args, spec_dict = self.argspec(self.body)
File "d:virtual_envspy364_32libsite-packagesinvoketasks.py", line 163, in argspec
raise TypeError("Tasks must have an initial Context argument!")
TypeError: Tasks must have an initial Context argument!
(py364_32) D:sourcespersonalpythonbplfabexamples>fab2 foo4 --arg1=3
Traceback (most recent call last):
File "d:virtual_envspy364_32libsite-packagesinvoketasks.py", line 160, in argspec
context_arg = arg_names.pop(0)
IndexError: pop from empty list
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "d:softwarepython364_32Librunpy.py", line 193, in _run_module_as_main
"__main__", mod_spec)
File "d:softwarepython364_32Librunpy.py", line 85, in _run_code
exec(code, run_globals)
File "D:virtual_envspy364_32Scriptsfab2.exe__main__.py", line 7, in <module>
File "d:virtual_envspy364_32libsite-packagesinvokeprogram.py", line 352, in run
self.parse_collection()
File "d:virtual_envspy364_32libsite-packagesinvokeprogram.py", line 444, in parse_collection
self.load_collection()
File "d:virtual_envspy364_32libsite-packagesfabric2main.py", line 87, in load_collection
super(Fab, self).load_collection()
File "d:virtual_envspy364_32libsite-packagesinvokeprogram.py", line 661, in load_collection
module, parent = loader.load(coll_name)
File "d:virtual_envspy364_32libsite-packagesinvokeloader.py", line 76, in load
module = imp.load_module(name, fd, path, desc)
File "d:virtual_envspy364_32libimp.py", line 235, in load_module
return load_source(name, filename, file)
File "d:virtual_envspy364_32libimp.py", line 172, in load_source
module = _load(spec)
File "<frozen importlib._bootstrap>", line 684, in _load
File "<frozen importlib._bootstrap>", line 665, in _load_unlocked
File "<frozen importlib._bootstrap_external>", line 678, in exec_module
File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
File "D:sourcespersonalpythonbplfabexamplesfabfile.py", line 59, in <module>
@wrap
File "d:virtual_envspy364_32libsite-packagesfabric2tasks.py", line 71, in task
return invoke.task(*args, **kwargs)
File "d:virtual_envspy364_32libsite-packagesinvoketasks.py", line 313, in task
return klass(args[0], **kwargs)
File "d:virtual_envspy364_32libsite-packagesfabric2tasks.py", line 21, in __init__
super(Task, self).__init__(*args, **kwargs)
File "d:virtual_envspy364_32libsite-packagesinvoketasks.py", line 77, in __init__
self.positional = self.fill_implicit_positionals(positional)
File "d:virtual_envspy364_32libsite-packagesinvoketasks.py", line 168, in fill_implicit_positionals
args, spec_dict = self.argspec(self.body)
File "d:virtual_envspy364_32libsite-packagesinvoketasks.py", line 163, in argspec
raise TypeError("Tasks must have an initial Context argument!")
TypeError: Tasks must have an initial Context argument!

似乎已经有一个与此 https://github.com/pyinvoke/invoke/issues/555 相关的未解决的问题,但到目前为止,我还没有找到解决方法。

你能解释一下我在这里做错了什么吗?但最重要的是,您能否提供一种方法来解决此 fabric2"限制"并能够创建一个装饰器,允许我使用 inspect 来捕获任务的名称以及在命令行中传递的参数?

提前谢谢。

自定义装饰器必须具有固定的初始上下文参数,如异常所述,因此func(c, *args, **kwargs)工作,而func(*args, **kwargs)则不起作用:

def trace(func):
@functools.wraps(func)
def wrapper(c, *args, **kwargs):
print(f'Begin {func.__name__}')
func(c, *args, **kwargs)
print(f'End {func.__name__}')
return wrapper
@task
@trace
def uname(c):
c.run('uname -a')

最新更新