裝飾器的應用場景

• 附加功能
• 資料的清理或新增:
  • 函式引數型別驗證 @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,**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)得到一個實際的裝飾器,帶引數的裝飾器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,**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,**kwargs):
    return self.run(*args,**kwargs)
  def run(self,**kwargs):
    raise NotImplementedError("子類未實現該介面")
def task(func):
  class SubTask(Task):
    def run(self,**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,**kwargs):
    self.args,self.kwargs = args,kwargs
    
  def __enter__(self):
    self.conn = pymysql.connect(*args,**kwargs)
    return self.conn.cursor()
    
  def __exit__(self,trackback):
    self.conn.close()

異常處理(避免重複)

傳播異常(__exit__中return False)

終止異常(__exit__中return True)

class BubleExceptions(object):
  def __enter__(self):
    return self
  def __exit__(self,trackback):
    if exc_instance:
      print("出現異常: {}".format(exc_instance)
    return False  # return True終止異常

處理特定的異常

class HandleValueError(object):
  def __enter__(self):
    return self
  def __exit__(self,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

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。