通過八個(gè) Python 裝飾器理解高階函數(shù)
1. 什么是高階函數(shù)?
高階函數(shù)是指至少滿足以下條件之一的函數(shù):
- 接受一個(gè)或多個(gè)函數(shù)作為參數(shù)。
- 返回一個(gè)函數(shù)。
裝飾器就是一種特殊的高階函數(shù),用來修改其他函數(shù)的行為。
2. 為什么使用裝飾器?
裝飾器可以讓我們?cè)诓恍薷脑己瘮?shù)代碼的情況下,增加額外的功能。比如記錄日志、計(jì)時(shí)、緩存等。這樣不僅讓代碼更簡(jiǎn)潔,也更容易維護(hù)。
3. 裝飾器的基本語(yǔ)法
裝飾器本質(zhì)上是一個(gè)接受函數(shù)作為參數(shù)的函數(shù),并返回一個(gè)新的函數(shù)。下面是一個(gè)簡(jiǎn)單的裝飾器示例。
def my_decorator(func):
def wrapper():
print("Something is happening before the function is called.")
func()
print("Something is happening after the function is called.")
return wrapper
@my_decorator
def say_hello():
print("Hello!")
say_hello()
輸出:
Something is happening before the function is called.
Hello!
Something is happening after the function is called.
4. 實(shí)戰(zhàn)案例六:參數(shù)驗(yàn)證裝飾器
參數(shù)驗(yàn)證裝飾器可以在執(zhí)行函數(shù)之前驗(yàn)證傳入?yún)?shù)的有效性。
def validate_params(*param_types):
def decorator(func):
def wrapper(*args, **kwargs):
if len(args) != len(param_types):
raise ValueError("Incorrect number of arguments")
for arg, param_type in zip(args, param_types):
if not isinstance(arg, param_type):
raise TypeError(f"Argument {arg} is not of type {param_type.__name__}")
return func(*args, **kwargs)
return wrapper
return decorator
@validate_params(int, int)
def multiply(x, y):
return x * y
print(multiply(3, 4)) # 正確
# print(multiply(3, "hello")) # 拋出 TypeError
輸出:
12
這個(gè)裝飾器會(huì)檢查傳入的參數(shù)類型是否符合預(yù)期。如果不符合,會(huì)拋出相應(yīng)的異常。
5. 實(shí)戰(zhàn)案例七:限制調(diào)用次數(shù)裝飾器
限制調(diào)用次數(shù)裝飾器可以在一定時(shí)間內(nèi)限制函數(shù)的調(diào)用次數(shù)。
import time
def limit_calls(max_calls, period):
def decorator(func):
call_times = []
def wrapper(*args, **kwargs):
current_time = time.time()
call_times.append(current_time)
while call_times and call_times[0] < current_time - period:
call_times.pop(0)
if len(call_times) > max_calls:
raise Exception("Too many calls in a short period of time")
return func(*args, **kwargs)
return wrapper
return decorator
@limit_calls(max_calls=2, period=5)
def process_request(data):
print(f"Processing request: {data}")
process_request("data1")
time.sleep(1)
process_request("data2")
time.sleep(1)
process_request("data3") # 拋出 Exception
輸出:
Processing request: data1
Processing request: data2
Too many calls in a short period of time
這個(gè)裝飾器限制了 process_request 函數(shù)在 5 秒內(nèi)只能被調(diào)用兩次。如果超過這個(gè)限制,會(huì)拋出異常。
6. 實(shí)戰(zhàn)案例八:緩存裝飾器(自定義)
自定義緩存裝飾器可以根據(jù)特定需求實(shí)現(xiàn)緩存功能。
from functools import wraps
def custom_cache(func):
cache = {}
@wraps(func)
def wrapper(*args, **kwargs):
key = (args, frozenset(kwargs.items()))
if key not in cache:
cache[key] = func(*args, **kwargs)
return cache[key]
return wrapper
@custom_cache
def fibonacci(n):
if n <= 1:
return n
else:
return fibonacci(n-1) + fibonacci(n-2)
for i in range(10):
print(fibonacci(i))
輸出:
0
1
1
2
3
5
8
13
21
34
這個(gè)裝飾器使用了一個(gè)字典 cache 來存儲(chǔ)函數(shù)的計(jì)算結(jié)果。如果相同的參數(shù)再次傳入,就直接從緩存中獲取結(jié)果。
7. 實(shí)戰(zhàn)案例九:異步裝飾器
異步裝飾器可以讓函數(shù)異步執(zhí)行,提高程序的響應(yīng)速度。
import asyncio
async def async_decorator(func):
async def wrapper(*args, **kwargs):
return await func(*args, **kwargs)
return wrapper
@async_decorator
async def download_file(url):
print(f"Downloading file from {url}")
await asyncio.sleep(2) # 模擬下載過程
print(f"File downloaded from {url}")
async def main():
await asyncio.gather(
download_file("http://example.com/file1"),
download_file("http://example.com/file2")
)
asyncio.run(main())
輸出:
Downloading file from http://example.com/file1
Downloading file from http://example.com/file2
File downloaded from http://example.com/file1
File downloaded from http://example.com/file2
這個(gè)裝飾器將函數(shù)轉(zhuǎn)換為異步函數(shù),并允許并行執(zhí)行多個(gè)任務(wù)。
8. 實(shí)戰(zhàn)案例十:多線程裝飾器
多線程裝飾器可以讓函數(shù)在多個(gè)線程中并發(fā)執(zhí)行。
import threading
def thread_decorator(func):
def wrapper(*args, **kwargs):
thread = threading.Thread(target=func, args=args, kwargs=kwargs)
thread.start()
return thread
return wrapper
@thread_decorator
def print_message(message):
print(message)
print_message("Hello, world!")
print_message("Goodbye, world!")
# 等待所有線程完成
for thread in threading.enumerate():
if thread != threading.current_thread():
thread.join()
輸出(順序可能不同):
Hello, world!
Goodbye, world!
這個(gè)裝飾器將函數(shù)放在不同的線程中執(zhí)行,并等待所有線程完成。