Python裝飾器的應(yīng)用場景代碼總結(jié)

更新時(shí)間：2020年04月10日 10:34:35 作者：臨淵

這篇文章主要介紹了Python裝飾器的應(yīng)用場景,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下

裝飾器的應(yīng)用場景

附加功能
數(shù)據(jù)的清理或添加:
- 函數(shù)參數(shù)類型驗(yàn)證 @require_ints 類似請(qǐng)求前攔截
- 數(shù)據(jù)格式轉(zhuǎn)換將函數(shù)返回字典改為 JSON/YAML 類似響應(yīng)后篡改
- 為函數(shù)提供額外的數(shù)據(jù) mock.patch
函數(shù)注冊(cè)
- 在任務(wù)中心注冊(cè)一個(gè)任務(wù)
- 注冊(cè)一個(gè)帶信號(hào)處理器的函數(shù)

不同應(yīng)用場景下裝飾器實(shí)現(xiàn)

函數(shù)注冊(cè)表

簡單注冊(cè)表

funcs = []
def register(func):
  funcs.append(func)
  return func    
@register
def a():
  return 3
  
@register
def b():
  return 5  
# 訪問結(jié)果
result = [func() for func in funcs]

注冊(cè)表隔離(使用類的不同實(shí)例)

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

執(zhí)行時(shí)封裝代碼

類型檢查

from functools import wraps
def require_ints(func):
  @wraps(func) # 將func的信息復(fù)制給inner
  def inner(*args, **kwargs):
    for arg list(args) + list(kwargs.values()):
      if not isinstance(arg, int:
        raise TypeError("{} 只接受int類型參數(shù)".format(func.__name__)
    return func(*args, **kwargs)
  return inner

用戶驗(yàn)證

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): # 將對(duì)象轉(zhuǎn)換為bool類型時(shí)調(diào)用
    return False
def requires_user(func):
  @wraps(func)
  def inner(user, *args, **kwargs): # 由于第一個(gè)參數(shù)無法支持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返回的字典格式轉(zhuǎn)為返回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("該條異常會(huì)被捕獲并按JSON格式輸出")

日志管理

import time
import logging
from functools import wraps

def logged(func):
  @wraps(func)
  def inner(*args, **kwargs): # *args可以裝飾函數(shù)也可以裝飾類
    start = time.time()
    result = func(*args, **kwargs)
    exec_time = time.time() - start
    logger = logging.getLoger("func.logged")
    logger.warning("{} 調(diào)用時(shí)間:{:.2} 執(zhí)行時(shí)間:{:.2}s 結(jié)果:{}".format(func.__name__, start, exec_time, result)

帶參數(shù)的裝飾器

帶參數(shù)的裝飾器相當(dāng)于一個(gè)返回裝飾器的函數(shù),@deco(a=1)在調(diào)用@之前會(huì)首先執(zhí)行deco(a=1)得到一個(gè)實(shí)際的裝飾器, 帶參數(shù)的裝飾器deco(a=1)模塊導(dǎo)入時(shí)立即執(zhí)行

裝飾類

為類增加可排序功能(而不通過繼承子類擴(kuò)充父類方法,比如多個(gè)類需要增加此功能時(shí))

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

也可定義一個(gè)SortableByCreated()類, 子類使用多重繼承其父類和SortableByCreated

類型轉(zhuǎn)換

函數(shù)被裝飾后有可能變?yōu)橐粋€(gè)類的實(shí)例,此時(shí)為了兼容函數(shù)調(diào)用,應(yīng)為所返回的類提供__call__方法

class Task(object):
  def __call__(self, *args, **kwargs):
    return self.run(*args, **kwargs)
  def run(self, *args, **kwargs):
    raise NotImplementedError("子類未實(shí)現(xiàn)該接口")
def task(func):
  class SubTask(Task):
    def run(self, *args, **kwargs):
      func(*args, **kwargs)
  return SubTask()

第二章上下文管理器

定義

包裝任意代碼

確保執(zhí)行的一致性

語法

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

應(yīng)用場景

資源清理

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()

異常處理(避免重復(fù))

傳播異常(__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("出現(xiàn)異常: {}".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的子類異常

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

更簡單的語法

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

以上就是本文的全部內(nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: