Python裝飾器的應用場景

裝飾器的應用場景

• 附加功能
• 數據的清理或添加:
  • 函數參數類型驗證 @require_ints 類似請求前攔截
  • 數據格式轉換 將函數返回字典改為 JSON/YAML 類似響應后篡改
  • 為函數提供額外的數據 mock.patch
• 函數注冊
  • 在任務中心注冊一個任務
  • 注冊一個帶信號處理器的函數

不同應用場景下裝飾器實現

函數注冊表

1. 簡單注冊表

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]

2. 注冊表隔離(使用類的不同實例)

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

執行時封裝代碼

1. 類型檢查

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

2. 用戶驗證

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

3. 輸出格式化

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

4. 異常捕獲

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格式輸出")

5. 日志管理

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)得到一個實際的裝飾器, 帶參數的裝飾器deco(a=1)模塊導入時立即執行

裝飾類

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 exec_type == ValueError則不處理ValueType的子類異常

也可以根據異常的屬性來判斷是否傳播或終止

更簡單的語法

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