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

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

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

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

簡(jiǎn)單注冊(cè)表

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

注冊(cè)表隔離(使用類(lèi)的不同實(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.registerdef a(): return 3 @r2.registerdef b(): return 5 @r1.register@r2.register

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

類(lèi)型檢查

from functools import wrapsdef 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類(lèi)型參數(shù)'.format(func.__name__) return func(*args, **kwargs) return inner

用戶(hù)驗(yàn)證

from functools import wrapsclass 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類(lèi)型時(shí)調(diào)用 return Falsedef requires_user(func): @wraps(func) def inner(user, *args, **kwargs): # 由于第一個(gè)參數(shù)無(wú)法支持self, 該裝飾器不支持裝飾類(lèi) if user and isinstance(user, User): return func(use, *args, **kwargs) else: raise ValueError('非合法用戶(hù)') return inner

輸出格式化

import jsonfrom functools import wrapsdef json_output(func): # 將原本func返回的字典格式轉(zhuǎn)為返回json字符串格式 @wrap(func) def inner(*args, **kwargs): return json.dumps(func(*args, **kwargs)) return inner

異常捕獲

import jsonfrom functools import wrapsclass 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_ouputdef error(): raise Error1('該條異常會(huì)被捕獲并按JSON格式輸出')

日志管理

import timeimport loggingfrom functools import wrapsdef logged(func): @wraps(func) def inner(*args, **kwargs): # *args可以裝飾函數(shù)也可以裝飾類(lèi) 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í)行

裝飾類(lèi)

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

import timefrom functools import wrapsdef 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()類(lèi), 子類(lèi)使用多重繼承其父類(lèi)和SortableByCreated

類(lèi)型轉(zhuǎn)換

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

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

第二章 上下文管理器

定義

包裝任意代碼

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

語(yǔ)法

with語(yǔ)句

__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)用場(chǎng)景

資源清理

import pymysqlclass 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...語(yǔ)句改為if exec_type == ValueError則不處理ValueType的子類(lèi)異常

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

更簡(jiǎn)單的語(yǔ)法

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

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。