Python装饰器与上下文管理器实战：从注册表到异常处理

程一诺

发布时间：2026-10-13 | 5人浏览过

原创

本文深入解析Python装饰器的多种应用场景，涵盖函数注册、参数类型验证、数据格式转换及日志管理等。同时详细阐述带参装饰器的执行机制与类装饰技巧，并系统介绍上下文管理器的定义、语法及在资源清理和异常传播控制中的具体实现，提供完整代码示例。

Python装饰器的核心应用场景

装饰器的应用场景

附加功能
数据的清理或添加:
- 函数参数类型验证：通过类似请求前拦截的机制，如@require_ints装饰器，在函数执行前检查传入参数是否符合预期的整数类型要求。
- 数据格式转换：在响应后篡改阶段，将函数原本返回的字典对象自动转换为JSON或YAML格式字符串，便于网络传输或存储。
- 为函数提供额外的数据：利用mock.patch等机制，在函数执行过程中注入模拟数据或替换依赖项，常用于测试环境。
函数注册
- 在任务中心注册一个任务：通过装饰器将普通函数标记为可执行任务，并统一纳入任务管理系统的调度池中。
- 注册一个带信号处理器的函数：将特定函数与系统信号关联，当信号触发时自动执行该函数，实现异步事件响应。

函数注册表的实现与隔离

不同应用场景下装饰器实现

函数注册表

简单注册表


funcs = []
def register(func):
  funcs.append(func)
  return func    
@register
def a():
  return 3
  
@register
def b():
  return 5  
# 访问结果
result = [func() for func in funcs]

注册表隔离：使用类的不同实例来管理各自的函数注册表，确保不同模块或组件间的注册函数互不干扰。


class Registry(object):
  def __init__(self):
    self._funcs = []
  
  def register(self, func):
    self._funcs.append(func)
    
  def run_all(self):
    return [func() for func in self._funcs] 
r1 = Registry()
r2 = Registry()

@r1.register
def a():
  return 3
  
@r2.register
def b():
  return 5
  
@r1.register
@r2.register

常见功能装饰器实现

执行时封装代码

类型检查


from functools import wraps
def require_ints(func):
  @wraps(func) # 将func的信息复制给inner
  def inner(*args, **kwargs):
    for arg list(args) + list(kwargs.values()):
      if not isinstance(arg, int:
        raise TypeError("{} 只接受int类型参数".format(func.__name__)
    return func(*args, **kwargs)
  return inner

用户验证


from functools import wraps

class User(object):
  def __init__(self, username, email):
    self.username = username
    self.email = email
    
class AnonymousUser(object):
  def __init__(self):
    self.username = self.email = None
  def __nonzero__(self): # 将对象转换为bool类型时调用
    return False
def requires_user(func):
  @wraps(func)
  def inner(user, *args, **kwargs): # 由于第一个参数无法支持self, 该装饰器不支持装饰类
    if user and isinstance(user, User):
      return func(use, *args, **kwargs)
    else:
      raise ValueError("非合法用户")
  return inner

输出格式化


import json
from functools import wraps
def json_output(func): # 将原本func返回的字典格式转为返回json字符串格式
  @wrap(func)
  def inner(*args, **kwargs):
    return json.dumps(func(*args, **kwargs))
  return inner

异常捕获


import json
from functools import wraps

class Error1(Exception):
  def __init__(self, msg):
    self.msg = msg
  def __str__(self):
    return self.msg
    

def json_output(func):
  @wrap(func)
  def inner(*args, **kwargs):
    try:
      result = func(*args, **kwargs)
    except Error1 as ex:
      result = {"status": "error", "msg": str(ex)}
    return json.dumps(result)
  return inner
# 使用方法
@json_ouput
def error():
  raise Error1("该条异常会被捕获并按JSON格式输出")

日志管理


import time
import logging
from functools import wraps

def logged(func):
  @wraps(func)
  def inner(*args, **kwargs): # *args可以装饰函数也可以装饰类
    start = time.time()
    result = func(*args, **kwargs)
    exec_time = time.time() - start
    logger = logging.getLoger("func.logged")
    logger.warning("{} 调用时间:{:.2} 执行时间:{:.2}s 结果:{}".format(func.__name__, start, exec_time, result)

带参装饰器与类装饰技巧

带参数的装饰器

带参数的装饰器相当于一个返回装饰器的函数。例如@deco(a=1)在调用@之前会首先执行deco(a=1)得到一个实际的装饰器。带参数的装饰器在模块导入时立即执行。

装饰类

为类增加可排序功能：无需通过继承子类扩充父类方法，例如当多个类需要增加此功能时，可直接使用装饰器动态修改类的行为。


import time
from functools import wraps
def sortable_by_created(cls):
  original_init = cls.__init__
  @wrap(original_init)
  def new_init(self, *args, **kwargs):
    original_init(*args, **kwargs)
    self._created = time.time()
  cls.__init__ = new_init
  
  cls.__lt__ = lambda self, other: self._created < other._created
  cls.__gt__ = lambda self, other: self._created > other._created
  return cls

也可定义一个SortableByCreated()类，子类使用多重继承其父类和SortableByCreated，以实现排序功能。

类型转换

函数被装饰后有可能变为一个类的实例，此时为了兼容函数调用，应为所返回的类提供__call__方法，使其可被直接调用。


class Task(object):
  def __call__(self, *args, **kwargs):
    return self.run(*args, **kwargs)
  def run(self, *args, **kwargs):
    raise NotImplementedError("子类未实现该接口")
def task(func):
  class SubTask(Task):
    def run(self, *args, **kwargs):
      func(*args, **kwargs)
  return SubTask()

上下文管理器：定义与语法

第二章上下文管理器

定义

包装任意代码

确保执行的一致性

语法

with语句

__enter__和__exit__方法


class ContextManager(object):
  def __init__(self):
    self.entered = False    
  def __enter__(self):
    self.entered = True
    return self    
  def __exit__(self, exc_type, exc_instance, traceback):
    self.entered = False

上下文管理器在异常处理中的应用

应用场景

资源清理


import pymysql

class DBConnection(object):
  def __init__(self, *args, **kwargs):
    self.args,self.kwargs = args, kwargs
    
  def __enter__(self):
    self.conn = pymysql.connect(*args, **kwargs)
    return self.conn.cursor()
    
  def __exit__(self, exc_type, exc_instance, trackback):
    self.conn.close()

异常处理：避免重复代码，通过上下文管理器统一处理资源释放逻辑。

传播异常：在__exit__中return False，允许异常继续向上传播。

终止异常：在__exit__中return True，表示异常已被处理，不再向外抛出。


class BubleExceptions(object):
  def __enter__(self):
    return self
  def __exit__(self, exc_type, exc_instance, trackback):
    if exc_instance:
      print("出现异常: {}".format(exc_instance)
    return False  # return True终止异常

处理特定的异常


class HandleValueError(object):
  def __enter__(self):
    return self
  def __exit__(self, exc_type, exc_instance, trackback):
    if not exc_type: return True
    if issubclass(exc_type, ValueError): 
      print("处理ValueError: {}".format(exc_instance)
    return False

将if issubclass...语句改为if exc_type == ValueError，则不处理ValueType的子类异常，实现更精确的异常控制。

也可以根据异常的属性来判断是否传播或终止，例如检查异常对象的特定字段。

更简单的语法：利用contextlib.contextmanager装饰器，以生成器形式简化上下文管理器的实现。


import contextlib

@contextlib.contextmanager
def acceptable_error_codes(*codes):
  try:
    yield
  except ShellException as exc_instance:
    if exc_instance.code not in codes:
      raise
    pass

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。