Python装饰器鉴赏：让代码更优雅的魔法
大约 12 分钟
Python装饰器鉴赏：让代码更优雅的魔法
作为一名有着五年测试开发经验的工程师，我深深被Python装饰器的优雅所折服。装饰器就像是给函数穿上了一件神奇的外衣，让我们能够在不修改原有代码的情况下，为函数添加各种强大的功能。
想象一下，你有一个普通的函数，但你希望为它添加日志记录、性能监控、权限验证等功能。传统的做法可能需要在每个函数内部添加大量重复代码，但装饰器让这一切变得简单而优雅。
装饰器的核心思想
装饰器的本质是一个接受函数作为参数并返回新函数的高阶函数。它遵循了"开闭原则"——对扩展开放，对修改封闭。这意味着我们可以在不改变原有函数代码的前提下，为其增加新的功能。
无参装饰器
def decorate(func):
def wrapper(*args, **kwargs):
# do something
ret = func(*args, **kwargs)
# do something
return ret
return wrapper
@decorate
def example():
print("hello world")在名为decorate的函数内定义一个闭包函数wrapper。wrapper捕获外部参数func函数，在func函数调用前后做一些前后处理，最终把func函数的结果返回出去。装饰器返回闭包函数wrapper的引用。这样就实现了一个简单的无参装饰器。
@只是python提供的一个装饰器调用方式的语法糖，@decorate def func和**func = decorate(func)**是等价的。
有参装饰器
def args_decorator(
my_argument
):
def decorator(func):
def wrapper(*args, **kwargs):
ret = func(*args, **kwargs)
print("decorator argument:{}".format(my_argument))
return ret
return wrapper
return decorator
@args_decorator(my_argument="lbwnb") # 添加带参数的装饰器 args_decorator()
def example():
print("call function example")
example()
# [output]:
# call function example
# decorator argument:lbwnb为了让装饰器可以带参数，需要在原装饰器外部再封装一层，最外层传入装饰器参数，内层传入函数的引用。
@args_decorator(my_argument="lbwnb") def func和 **func = args_decorator(my_argument="lbwnb")(func)**是等价的。
装饰器类
python 类的使用(5)之类装饰器(类的装饰器和类作为装饰器)_类装饰器可以设定条件来决定使不使用吗-CSDN博客
相当于自己实现了__call__
class Tracer():
def __init__(self, func):
self.func = func
self.calls = 0
def __call__(self, *args, **kwargs):
self.calls += 1
print("call %s() %d times" % (self.func.__name__, self.calls))
return self.func(*args, **kwargs)
@Tracer
def test_tracer():
print("test trace")
for i in range(3):
test_tracer()
# [output]
# call test_tracer() 1 times
# test trace
# call test_tracer() 2 times
# test trace
# call test_tracer() 3 times
# test trace装饰器类和装饰器函数类似，都可以对函数实现前后处理。
装饰器鉴赏
内置装饰器
@staticmethod
相信大家都用过staticmethod装饰器，这个装饰器可以对类的方法进行修饰，被修饰的方法称为静态方法。访问该方法（其实应该说是函数）时不需要有对应的实例和类。
class A:
@staticmethod
def test():
print("hello world")
def test1(self):
print("test1")
a = A()
A.test()
a.test1()
print(A.test)
print(a.test1)
# [output]
# hello world
# test1
# <function A.test at 0x102342940>
# <bound method A.test1 of <__main__.A object at 0x102520fa0>>可以看到，test静态方法实际上是个函数function，而test1实例方法实际上是方法method。
staticmethod源码使用c语言编写，用python实现也非常简单，下面提供一个python版本：
class MyStaticmethod(object):
def __init__(self, func):
self.func = func
def __get__(self, obj, objtype=None):
return self.func
class A:
@MyStaticmethod
def test(): #代码编辑器有可能标红
print("hello world")
A.test()
print(A.test)
# [output]
# hello world
# <function A.test at 0x10f753700>定义一个装饰器类，但我们不需要重新更改__call__方法，因为我们不需要做相应的前后处理。我们仅仅通过重写了__get__描述符，在获取类属性的时候返回自己接收到的func函数即可。观察输出可以看到，test静态方法确实变成了函数而不是方法。
@classmethod
classmethod装饰器修饰的方法被称为类方法。类方法不需要传递实例的引用，但是需要传递类的引用。classmethod底层也是由c代码编写，其实现逻辑与staticmethod类似。下面是python代码的模拟实现。
class MyClassMethod(object):
def __init__(self, func):
self.func = func
def __get__(self, obj, obj_class=None):
if obj_class is None:
obj_class = type(obj)
return MethodType(self.func,obj_class)
class A:
@MyClassMethod
def test(cls):
print("hello world:{}".format(cls))
def test1(self):
print("test1")
a = A()
A.test()
a.test1()
print(A.test)
print(a.test1)
# [output]:
# hello world:<class '__main__.A'>
# test1
# <bound method A.test of <class '__main__.A'>>
# <bound method A.test1 of <__main__.A object at 0x100f56d60>>第三方装饰器
@wraps
装饰器的副作用之一就是函数使用了装饰器后，函数的原信息也丢失了。比如上述的无参装饰器，对于被其装饰的函数，用户最终拿到的其实是wrapper函数而不是原函数。
def decorator1(func):
print('func address: [{}]'.format(func))
def wrapper(*args, **kwargs):
print('wrapper address: [{}]'.format(wrapper))
ret = func(*args, **kwargs)
return ret
return wrapper
@decorator1
def example():
print("call function [{}]--[{}]".format(example.__name__, example))
example()
# [output]:
# func address: [<function example at 0x1004d0ca0>]
# wrapper address: [<function decorator1.<locals>.wrapper at 0x1005c2700>]
# call function [wrapper]--[<function decorator1.<locals>.wrapper at 0x1005c2700>]我们对wrapper函数加上 **wraps装饰器 **再看看。
def decorator2(func):
print('func address: [{}]'.format(func))
@wraps(func)
def wrapper(*args, **kwargs):
print('wrapper address: [{}]'.format(wrapper))
ret = func(*args, **kwargs)
return ret
return wrapper
@decorator2
def example():
print("call function [{}]--[{}]".format(example.__name__, example))
example()
# [output]:
# func address: [<function example at 0x10ad68ca0>]
# wrapper address: [<function example at 0x10ae59700>]
# call function [example]--[<function example at 0x10ae59700>]通过函数地址看，最终拿到的函数仍然是wrapper函数，但是函数信息却是func的信息。
wraps装饰器通过核心的update_wrapper函数，将原函数的信息更新到wrapper上。
def update_wrapper(wrapper,
wrapped,
assigned = WRAPPER_ASSIGNMENTS,
updated = WRAPPER_UPDATES):
"""Update a wrapper function to look like the wrapped function
wrapper is the function to be updated
wrapped is the original function
assigned is a tuple naming the attributes assigned directly
from the wrapped function to the wrapper function (defaults to
functools.WRAPPER_ASSIGNMENTS)
updated is a tuple naming the attributes of the wrapper that
are updated with the corresponding attribute from the wrapped
function (defaults to functools.WRAPPER_UPDATES)
"""
for attr in assigned:
try:
value = getattr(wrapped, attr)
except AttributeError:
pass
else:
setattr(wrapper, attr, value)
for attr in updated:
getattr(wrapper, attr).update(getattr(wrapped, attr, {}))
wrapper.__wrapped__ = wrapped
return wrapper@lru_cache
被lru_cache装饰器装饰的函数，可以缓存其函数结果，后续使用相同参数调用该函数时，不需要执行函数，直接调用缓存结果。缓存规则使用LRU（最近最少使用）算法。
@lru_cache()
def m_sum(*args):
print("go into sum")
return sum(args)
print(m_sum(1, 2, 3))
print(m_sum(1, 2, 3))
# [output]
# go into sum
# 6
# 6lru_cache的核心代码如下：
def wrapper(*args, **kwds):
# Size limited caching that tracks accesses by recency
nonlocal root, hits, misses, full
key = make_key(args, kwds, typed)
with lock:
link = cache_get(key)
if link is not None:
# Move the link to the front of the circular queue
link_prev, link_next, _key, result = link
link_prev[NEXT] = link_next
link_next[PREV] = link_prev
last = root[PREV]
last[NEXT] = root[PREV] = link
link[PREV] = last
link[NEXT] = root
hits += 1
return result
misses += 1
result = user_function(*args, **kwds)
with lock:
if key in cache:
# Getting here means that this same key was added to the
# cache while the lock was released. Since the link
# update is already done, we need only return the
# computed result and update the count of misses.
pass
elif full:
# Use the old root to store the new key and result.
oldroot = root
oldroot[KEY] = key
oldroot[RESULT] = result
# Empty the oldest link and make it the new root.
# Keep a reference to the old key and old result to
# prevent their ref counts from going to zero during the
# update. That will prevent potentially arbitrary object
# clean-up code (i.e. __del__) from running while we're
# still adjusting the links.
root = oldroot[NEXT]
oldkey = root[KEY]
oldresult = root[RESULT]
root[KEY] = root[RESULT] = None
# Now update the cache dictionary.
del cache[oldkey]
# Save the potentially reentrant cache[key] assignment
# for last, after the root and links have been put in
# a consistent state.
cache[key] = oldroot
else:
# Put result in a new link at the front of the queue.
last = root[PREV]
link = [last, root, key, result]
last[NEXT] = root[PREV] = cache[key] = link
# Use the cache_len bound method instead of the len() function
# which could potentially be wrapped in an lru_cache itself.
full = (cache_len() >= maxsize)
return result整个lru缓存算法的核心数据结构是字典cache和存储热点信息的双向循环链表。
cache的键是函数参数hash值，值是双向循环链表的节点。
以上述的例子m_sum函数为例：
- 先对m_sum函数的参数args进行hash计算（所以该装饰器也限制了传入函数的参数必须是可以被hash的）得到key。
- 如果缓存字典里有这个key，拿到对应的链表节点，将该节点放置链表末尾（即认为该节点是最近最多使用的节点），最后直接返回这个节点存储的result值；若不存在这个key，直接调用函数得到结果，并记为一次缓存为命中。
- lru缓存装饰器做了并发逻辑判断，防止m_sum函数调用期间，其他线程更新了m_sum的lru缓存，重复更新节点导致链表状态异常。在m_sum函数调用后，继续判断参数args的key是否在缓存字典中。
- 若在缓存字典里，则无需更新链表，因为其他线程已经更新了。若不在缓存字典里，则根据lru的替换规则对链表和缓存字典做更新。
@singledispatch
Python无法直接实现函数重载。通过functools包里的singledispatch装饰器，可以让函数变成'泛函数'，模拟实现类似java和c++的函数重载。
@singledispatch
def get_port(obj):
pass
# 参数字符串
@get_port.register(str)
def _(port):
print("get port", port, type(port), "str")
# 参数int
@get_port.register(int)
def _(n):
print("get port", n, type(n), "int")
get_port(80)
get_port("80")
# [output]
# get port 80 <class 'int'> int
# get port 80 <class 'str'> str下面是singledispatch核心代码（完整版可以去看源码）
def singledispatch(func):
...
registry = {}
def dispatch(cls):
...
try:
impl = registry[cls]
except KeyError:
impl = _find_impl(cls, registry)
return impl
def register(cls, func=None):
...
registry[cls] = func
...
return func
def wrapper(*args, **kw):
if not args:
raise TypeError(f'{funcname} requires at least '
'1 positional argument')
return dispatch(args[0].__class__)(*args, **kw)
...
wrapper.register = register
wrapper.dispatch = dispatch
return wrapper被singledispatch修饰过的函数拥有register方法。通过register方法注册不同类型的参数函数，最后在函数调用时根据参数类型进行查找分派。
单分派装饰器利用了单参数类型的唯一性，实现了方法分派。仿照单分派函数，multimethod库实现了对函数参数的多个类型组合成可哈希对象（tuple或str），建立起不同类型参数的唯一性，从而实现多分派函数。
https://pypi.org/project/multimethod/
@dataclass
dataclass可以将对象转换为数据类型，方便开发者进行数据处理。最常用的方式是用dataclass减少初始化函数的繁琐代码。
from dataclasses import dataclass
@dataclass
class InventoryItem:
"""Class for keeping track of an item in inventory."""
name: str
unit_price: float
quantity_on_hand: int = 0
# 等价于
class InventoryItem:
def __init__(self, name: str, unit_price: float, quantity_on_hand: int = 0):
self.name = name
self.unit_price = unit_price
self.quantity_on_hand = quantity_on_handdataclass通过_set_new_attribute方法动态生成__init__方法，并在__init__方法实现类成员变量-》实例变量的赋值。
if init:
# Does this class have a post-init function?
has_post_init = hasattr(cls, _POST_INIT_NAME)
# Include InitVars and regular fields (so, not ClassVars).
flds = [f for f in fields.values()
if f._field_type in (_FIELD, _FIELD_INITVAR)]
_set_new_attribute(cls, '__init__',
_init_fn(flds,
frozen,
has_post_init,
# The name to use for the "self"
# param in __init__. Use "self"
# if possible.
'__dataclass_self__' if 'self' in fields
else 'self',
globals,
))@app.route
在搭建flask服务时，我们需要对视图函数做路由规则映射，经常需要使用到app.route装饰器。
@app.route("/")
def index():
return "Hello, World!"下面看到route装饰器的源码
@setupmethod
def route(self, rule: str, **options: t.Any) -> t.Callable[[T_route], T_route]:
"""Decorate a view function to register it with the given URL
rule and options. Calls :meth:`add_url_rule`, which has more
details about the implementation.
.. code-block:: python
@app.route("/")
def index():
return "Hello, World!"
See :ref:`url-route-registrations`.
The endpoint name for the route defaults to the name of the view
function if the ``endpoint`` parameter isn't passed.
The ``methods`` parameter defaults to ``["GET"]``. ``HEAD`` and
``OPTIONS`` are added automatically.
:param rule: The URL rule string.
:param options: Extra options passed to the
:class:`~werkzeug.routing.Rule` object.
"""
def decorator(f: T_route) -> T_route:
endpoint = options.pop("endpoint", None)
self.add_url_rule(rule, endpoint, f, **options)
return f
return decoratorroute装饰器把装饰器入参里的路由规则'rule'通过self.add_url_rule方法注册到自己的url_map中，url_map存储了{url:endpoint}(可以理解为函数名字)的映射，同时建立一个{endpoint:view_functions}映射。在请求到来时，dispatch_request函数会进行请求分发，通过请求的url找到对应的endpoint，再从endpoint找到view_functions，最后执行该函数。
def dispatch_request(self) -> ft.ResponseReturnValue:
"""Does the request dispatching. Matches the URL and returns the
return value of the view or error handler. This does not have to
be a response object. In order to convert the return value to a
proper response object, call :func:`make_response`.
.. versionchanged:: 0.7
This no longer does the exception handling, this code was
moved to the new :meth:`full_dispatch_request`.
"""
req = request_ctx.request
if req.routing_exception is not None:
self.raise_routing_exception(req)
rule: Rule = req.url_rule # type: ignore[assignment]
# if we provide automatic options for this URL and the
# request came with the OPTIONS method, reply automatically
if (
getattr(rule, "provide_automatic_options", False)
and req.method == "OPTIONS"
):
return self.make_default_options_response()
# otherwise dispatch to the handler for that endpoint
view_args: dict[str, t.Any] = req.view_args # type: ignore[assignment]
return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)DIY装饰器
字典参数检查装饰器
有时候为了灵活的函数传参，我们把参数都包在一个字典里，将这个字典作为输入。这样的副作用是参数定义不清晰，并且如果字典里没有对应参数，会在函数运行中报错。
通过下面的参数校验装饰器，可以标识出参数定义，还可以在进入函数前做参数校验。
class DemoRunner(TestRunner):
@StepInputConfig([
StepParam(key="name", type=str, require=True, note="姓名"),
StepParam(key="age", type=int, require=True, note="年龄"),
StepParam(key="country", type=str, require=False, default="China", note="国籍")
])
def hello(self, step_context):
step_input = step_context.step_input
name = step_input["name"]
age = step_input["age"]
country = step_input["country"]
self.logger.info("%s, %d years old, from %s", name, age, country)
class StepInputConfig(JsonableClass):
def __init__(self, step_configs):
# type: (List[StepParam]) -> None
self.step_configs = step_configs
def __call__(self, func):
@wraps(func)
def _step_rapper(runner, step_context):
for param in self.step_configs:
step_input = step_context.step_input
if param.key not in step_input:
if param.require:
raise Exception("Required step input '%s' not found!" % param.key)
else:
step_input[param.key] = param.default
else:
if not isinstance(step_input[param.key], param.type):
raise Exception(
"'%s' step input type error, expect: %s, actually: %s." % (
param.key, param.type, type(step_input[param.key])))
return func(runner, step_context)
return _step_rapper函数结果缓存装饰器
类似于lru_cache，这段代码实现了一个简易的函数结果缓存装饰器。
class CallCache(object):
def __init__(self):
self.cache = {}
def __call__(self, func):
@wraps(func)
def callee(): # notice: do not support functions with arguments by design, for prevent cache flood DoS.
key = func.__name__
if key in self.cache:
result = self.cache.get(key)
else:
result = func()
self.cache[key] = result
return result
return callee
callcache = CallCache()还有很多经典的装饰器，以下内容由chatgpt提供。
鉴权装饰器
def authentication_required(min_role):
def decorator(func):
def wrapper(user, *args, **kwargs):
if user and hasattr(user, 'role') and user.role >= min_role:
# 用户已认证且权限足够，执行原函数
return func(user, *args, **kwargs)
else:
# 用户未认证或权限不足，返回错误信息或执行其他操作
return "Authentication failed. Insufficient permissions."
return wrapper
return decorator
class User:
def __init__(self, role):
self.role = role
@authentication_required(min_role=2)
def sensitive_operation(user):
return f"Sensitive operation executed for user with role {user.role}."调试信息装饰器
def debug_decorator(func):
def wrapper(*args, **kwargs):
print(f"Function {func.__name__} called with arguments: {args}, {kwargs}")
result = func(*args, **kwargs)
print(f"Function {func.__name__} returned: {result}")
return result
return wrapper
# 使用示例
@debug_decorator
def add(a, b):
return a + b
print(add(3, 5))函数计时装饰器
import time
def timing_decorator(func):
def wrapper(*args, **kwargs):
start_time = time.time()
result = func(*args, **kwargs)
end_time = time.time()
print(f"Function {func.__name__} executed in {end_time - start_time:.4f} seconds.")
return result
return wrapper
# 使用示例
@timing_decorator
def some_function():
time.sleep(2)
print("Function executed.")
some_function()