functools 【可調用物件】
簡介
functools module 用於簡化任務的工具箱，提供了一系列工具來以函數/方法式編程風格處理函數。
@cache
@functools.cache(user_function) 用於向函數和方法新增記憶功能 (Memoization)。當相同的輸入再次出現時，函數的結果將被存儲和重用。 這可以提高經常使用相同參數的函數的調用性能。 @functools.cache 等效 @functools.lru_cache(maxsize=None, typed=False)
@functools.cache 等效 @functools.lru_cache(maxsize=None, typed=False)這會使用無限的 cache，並且不會刪除舊值 (不會主動刪除舊 cache)。
對於僅使用有限參數的函數的調用，會比 @functools.lru_cache 更節省 memory 且速度更快。 但是，請注意唯一參數的數量，@functools.cache 可能會填滿您的 memory。
範例 – 費波那契數, 測量時間
@cache 包裝了 fibonacci() 遞迴函數，先前計算的結果將被 cache 和重用，從而顯著加快計算速度。 如果沒有 cache，遞迴函數將進行大量的重複計算。
首先測量執行時間。之後，Python 將檢查給定參數的函數結果是否在 cache 中。 如果是，Python 返回 cache 的結果； 如果不是，Python 計算後，將結果存儲在 cache 中。
在此範例中，@functools.cache 必須為第一個 decorator (因此最後寫入)。
因為測量函數的實際執行時間更有意義，而不是返回 @cache 的結果。
時間複雜度表示法 (例如：O(a*b) 或 O(2^n)):
為我們提供了一個高級的、概括的概念，即算法的運行時間如何隨著輸入大小的增加而增加。 但它並沒有直接告訴我們算法的絕對運行速度有多快，因為這還取決於算法中執行的特定 operations。
如範例，Fibonacci sequence 函數正在遞迴計算 35 個斐波那契數。 如果沒有記憶性，這將涉及大量的冗餘計算。 當函數被記憶時，它只需要計算每個 Fibonacci sequence 一次。因此，雖然 Fibonacci sequence 的原始時間複雜度確實是 O(2^n)，但記憶後的時間複雜度是 O(n)。
import time
import functools
def measure_execution_time(func):
@functools.wraps(func)
def wrapper(*args, _is_recursive_call=False, **kwargs):
if not _is_recursive_call:
wrapper._start_time = time.time() # Set the start time only for the top-level call
result = func(*args, **kwargs) # Call the function
if not _is_recursive_call: # If this is the top-level call
wrapper._end_time = time.time() # Set the end time
print(f"{func.__name__} executed in {wrapper._end_time - wrapper._start_time:.3f} s.")
return result
return wrapper
@functools.cache
@measure_execution_time
def fibonacci_1(n): # Fibonacci sequence (O(2^n))
if n < 2:
return n
return fibonacci_1(n-1, _is_recursive_call=True) + fibonacci_1(n-2, _is_recursive_call=True)
@measure_execution_time
def fibonacci_2(n): # Fibonacci sequence (O(2^n))
if n < 2:
return n
return fibonacci_2(n-1, _is_recursive_call=True) + fibonacci_2(n-2, _is_recursive_call=True)
fibonacci_1(20) # Output: 0.000 s
fibonacci_1(35) # Output: 0.000 s
fibonacci_2(20) # Output: 0.004 s
fibonacci_2(35) # Output: 5.472 s範例 – 昂貴操作
expcious_operation() 的結果將根據參數 a 和 b 進行 cache。 如果你使用相同的參數再次調用 expcious_operation()，Python 不會重新計算該函數，而是直接返回 cache 結果。
expense_operation() 函數運行兩個 nested loops， 這會導致執行一億次 iterations，每次 iteration 都涉及一次賦值操作。(記憶後的時間複雜度仍然維持 O(a*b))
import functools
@functools.cache
def expensive_operation(a, b):
print('[Simulating an expensive operation O(a*b)]')
for i in range(a):
for j in range(b):
x = i
y = j
return a*b
print(expensive_operation(10000, 10000)) # Output: 1000000
print(expensive_operation(10000, 10000)) # Output: 1000000範例 – 遞迴 vs 迴圈
import time
import functools
def measure_execution_time(func):
@functools.wraps(func)
def wrapper(*args, _is_recursive_call=False, **kwargs):
if not _is_recursive_call:
wrapper._start_time = time.time() # Set the start time only for the top-level call
result = func(*args, **kwargs) # Call the function
if not _is_recursive_call: # If this is the top-level call
wrapper._end_time = time.time() # Set the end time
print(f"{func.__name__} executed in {wrapper._end_time - wrapper._start_time:.3f} s.")
return result
return wrapper
@functools.cache
@measure_execution_time
def fibonacci_recursion(n): # Fibonacci sequence O(2^n)
if n < 2:
return n
return fibonacci_recursion(n-1, _is_recursive_call=True) + fibonacci_recursion(n-2, _is_recursive_call=True)
@functools.cache
@measure_execution_time
def fibonacci_for(n): # Fibonacci sequence O(n)
a, b = 0, 1 # Initialize
for _ in range(n):
a, b = b, a + b
return a
@functools.cache
@measure_execution_time
def fibonacci_while(n): # Fibonacci sequence O(n)
a, b = 0, 1 # Initialize
while n > 0:
a, b = b, a + b
n -= 1
return a
fibonacci_recursion(20) # Output: 0.000 s
fibonacci_recursion(35) # Output: 0.000 s
fibonacci_for(20) # Output: 0.000 s
fibonacci_for(35) # Output: 0.000 s
fibonacci_while(20) # Output: 0.000 s
fibonacci_while(35) # Output: 0.000 s
@lru_cache
提供 LRU(Least Recently Used；最近最少使用) cache 的功能。 如果 cache 已滿，它將丟棄最近最少使用的項目。
lru_cache(maxsize=128, typed=False)maxsize：cache 最大的大小。 最多可保存maxsize個最近調用的 cache。如果設置為 None，則cache可以無限增長。typed：如果等於True，不同類型的參數將被單獨 cache。 例如：f(3),f(3.0)將被視為不同調用(int, float)。f(('answer', 3)),f(('answer', 3.0))將被視為相同調用(Tuple)。
範例 – Argument 類型差異
import functools
@functools.lru_cache(maxsize=128, typed=True)
def add(x, y):
return x + y
add(3, 4) # Will be cached
print(add.cache_info())
add(3.0, 4.0) # Will be cached (typed=True)
print(add.cache_info())
add(x=3.0, y=4.0) # Will be cached (argument patterns)
print(add.cache_info())
add(y=4.0, x=3.0) # Will be cached (argument patterns)
print(add.cache_info())
add(x=3.0, y=4.0)
print(add.cache_info())
add.cache_clear() # Clear the cache
print(add.cache_info()) 執行結果：
CacheInfo(hits=0, misses=1, maxsize=128, currsize=1)
CacheInfo(hits=0, misses=2, maxsize=128, currsize=2)
CacheInfo(hits=0, misses=3, maxsize=128, currsize=3)
CacheInfo(hits=0, misses=4, maxsize=128, currsize=4)
CacheInfo(hits=1, misses=4, maxsize=128, currsize=4)
CacheInfo(hits=0, misses=0, maxsize=128, currsize=0)範例 – Thread-safe
decorator 預設情況下不是 Thread-safe。 在 multi-threaded 環境中，不同的 threads 可能會嘗試同時訪問或修改 cache，從而導致競爭條件，不可預測的結果。
因此，在此類似的環境中使用 decorator 時，應始終確保 Thread-safe。
在 multi-threaded 環境中使用 @lru_cache 時，確保 function operation, cache management 都是 thread-safe 的非常重要。
import functools
import threading
from concurrent.futures import ThreadPoolExecutor
# Define a lock
lock = threading.Lock()
# Time-consuming operation
def expensive_operation(n):
return n * n
# Thread safety measures
@functools.lru_cache
def thread_safe_expensive_operation(n):
with lock:
return expensive_operation(n)
# Using ThreadPoolExecutor
with ThreadPoolExecutor() as executor:
# Submit tasks
futures = [executor.submit(thread_safe_expensive_operation, i) for i in range(35)]
# Print results
for future in futures:
print(future.result())import functools
import threading
from concurrent.futures import ThreadPoolExecutor
# Define a lock
lock = threading.Lock()
# Fibonacci sequence
def fibonacci(n):
if n < 2:
return n
return fibonacci(n-1) + fibonacci(n-2)
# Thread safety measures
@functools.lru_cache
def thread_safe_expensive_operation(n):
with lock:
return fibonacci(n)
# Using ThreadPoolExecutor
with ThreadPoolExecutor() as executor:
# Submit tasks
futures = [executor.submit(thread_safe_expensive_operation, i) for i in range(35)]
# Print results
for future in futures:
print(future.result())Non-thread safe：
import functools
from concurrent.futures import ThreadPoolExecutor
# Time-consuming operation
@functools.lru_cache
def fibonacci_recursion(n): # Fibonacci sequence
if n < 2:
return n
return fibonacci_recursion(n-1) + fibonacci_recursion(n-2)
# Using ThreadPoolExecutor
with ThreadPoolExecutor() as executor:
# Submit tasks
futures = [executor.submit(fibonacci_recursion, i) for i in range(35)]
# Print results
for future in futures:
print(future.result())cache_info()
cache_info() 提供有關 cache 使用情況的報告(非 memory 的報告)。
包含：
hits：在 cache 中找到所需結果的次數。 表示有hits次不需要計算該函數，因為結果已經從之前的計算中獲得。misses：在 cache 中找不到所需結果，表示調用函數的次數。maxsize：cache 可以容納的最大大小，表示最多可以存儲maxsize個結果。currsize：當前 cache 的大小，表示已經存儲currsize個結果。
import time
import functools
@functools.lru_cache(maxsize=128, typed=False)
def fib(n): # Fibonacci sequence
if n < 2:
return n
return fib(n-1) + fib(n-2)
print(fib(35))
print(fib.cache_info())
# Output: CacheInfo(hits=33, misses=36, maxsize=128, currsize=36)cache_clear()
cache_clear() 用於手動清除所有 cache。 當您知道函數的結果可能會更改 (例如：結果取決於外部資源或狀態)，或者您想要釋放 memory 時，這非常有用。
import functools
@functools.cache
def expensive_operation(a, b):
print('[Simulating an expensive operation]')
return a**b
print(expensive_operation(2, 3))
print(expensive_operation(2, 3)) # Reusing old results
print(expensive_operation(3, 3))
expensive_operation.cache_clear()
print(expensive_operation(2, 3))執行結果：
[Simulating an expensive operation]
8
8
[Simulating an expensive operation]
27
[Simulating an expensive operation]
8範例 – 外部 Data source 交互
考慮一個與某些外部 data source 交互的函數(例如：數據庫、文件或 API)。 如果該 data source 發生更改，您需要確保沒有使用已過時的 cached 結果。
import time
import functools
# Assume data source
database = {"key": 50}
@functools.lru_cache(maxsize=128)
def get_data(key):
time.sleep(1) # Simulate delay(remote database)
return database[key]
# Fetches data
print(get_data("key")) # Output: 50
# Data changes
database["key"] = 100
print(get_data("key")) # Output: 50
# Clear the cache
get_data.cache_clear()
print(get_data("key")) # Output: 100@cached_property
@cached_property(func) 用於 Class 中的方法新增記憶功能，並轉換為屬性(即，無需括號即可訪問， 計算出來的值與 Class 實例相關聯)。
該屬性的值在首次訪問時計算一次，然後在 Class 實例的 lifetime (生命週期) 內，作為一般屬性進行 cached。這使得 @cached_property 適合用於與實例相關的昂貴計算。
它的工作原理如下：
首次訪問某個屬性時，Python 會檢查該名稱的屬性是否已存在。
如果不存在，則調用
@cached_property方法，並將結果存儲為同名的屬性。之後，cached 屬性的行為就像一般屬性一樣。 任何後續的讀取都將獲取 cached 值，任何寫入都將覆蓋 cached 值。
如果要清除 cached 值，應該使用
del刪除該屬性。 下次訪問該屬性時，@cached_property將再次運行。
@cached_property 是針對每個實例的，而不是對每個 Clas 的。(Clas 的每個實例都將擁有自己的屬性 cached)
因此如果您有多個 Clas 實例，它們可能會使用大量 memory。
Cached_property vs. Cache
你應該兩者都使用嗎？ 在同一 Class 方法中，同時使用 @cached_property, @cache 通常是不必要的，並且可能會導致混亂。 這是因為它們是針對不同的場景而設計的：@cached_property 用於計算屬性值的方法，@cache 用於使用輸入，執行計算的函數。 然而，在某些情況下，使用兩者可能是有益的，例如：具有計算成本昂貴的 Class 方法，在內部調用另一個計算成本昂貴的函數。
總之，使用哪一種主要取決於您的具體用例、函數/方法的計算強度、函數/方法是否使用可變數據或有副作用、您環境的內存限制以及您的參數是否 函數/方法是可散列的。
@cached_property 優點：
當屬性計算量很大並且屬性被多次訪問時，它可以加速您的程序。
作為一個屬性，它不需要顯式調用函數或方法，從而使您的程式碼更清晰、更具可讀性。
目的用於建立唯讀屬性(請避免修改)。 cached 結果存儲在實例變數中。
@cache 優點：
當函數計算量很大並且使用相同的參數多次調用該函數，它可以加速您的程序。
它可以 cached 具有任意數量參數的函數的結果。
@cached_property 缺點：
它只能與不帶參數的實例方法一起使用(self 除外)。
@cache 缺點：
它只能與接受 Hashable 參數的函數/方法一起使用，因為 cache 是根據 dictionary 實現的。
共同缺點：
如果函數有依賴於 mutable 數據，cache 可能容易導致不正確的結果。(將繼續返回舊的 cached 值)
它使用 memory 來存儲 cache，這在大數據或 memory 有限的環境中可能會出現問題。
from functools import cache, cached_property
class MyClass():
@cache
def complex_operation1(self):
print("Performing complex operation...")
result = 123 # 假設是複雜的 operation
return result
@cached_property
def complex_operation2(self):
print("Performing complex operation...")
result = 123 # 假設是複雜的 operation
return result
# Create an instance
obj = MyClass()
print(obj.complex_operation1()) # Output: Performing complex operation... 123
print(obj.complex_operation1()) # Output: 123
print(obj.complex_operation2) # Output: Performing complex operation... 123
print(obj.complex_operation2) # Output: 123範例 – 基礎
import math
from functools import cached_property
class Circle:
def __init__(self, radius):
self.radius = radius
@cached_property
def area(self):
print("Calculating area...")
return math.pi * (self.radius ** 2)
c = Circle(5)
print(c.area) # Outputs: Calculating area... and then 78.53981633974483
print(c.area) # Outputs: 78.53981633974483
from functools import cached_property
class MyClass:
@cached_property
def complex_operation(self):
print("Performing complex operation...")
result = 123 # 假設是複雜的 operation
return result
# Create instance
obj = MyClass()
# 第 1 次訪問，執行 complex_operation()
print(obj.complex_operation) # Output: Performing complex operation... 123
# 第 2 次訪問，訪問 cached 值
print(obj.complex_operation) # Output: 123
# Overwrite cached value
obj.complex_operation = 456
print(obj.complex_operation) # Output: 456
# Clear cached value
del obj.complex_operation
# 第 1 次訪問，執行 complex_operation()
print(obj.complex_operation) # Output: Performing complex operation... 123範例 – Thread-safe
@functools.cached_property 是 thread-safe，這意味著它可以在 multithreaded 環境中安全使用。
如果沒有 thread-safe，一個 thread 可能會看到該屬性處於部分更新狀態，而另一個 thread 正在更新該屬性。 (看不到兩者之間的某些不一致狀態)
import time
import threading
import concurrent.futures
from functools import cached_property
class ExpensiveObject:
def __init__(self):
self._expensive_data = None
@cached_property
def expensive_data(self):
print("Performing complex operation...")
time.sleep(2)
return sum(i * i for i in range(10000)) # 假設是複雜的 operation
def worker(obj):
print(f"Thread {threading.current_thread().name}: {obj.expensive_data}")
if __name__ == "__main__":
obj = ExpensiveObject()
with concurrent.futures.ThreadPoolExecutor(thread_name_prefix="WorkerThread", max_workers=3) as executor:
for _ in range(5):
executor.submit(worker, obj)執行結果：
Performing complex operation...
Thread WorkerThread_0: 333283335000
Thread WorkerThread_1: 333283335000
Thread WorkerThread_0: 333283335000
Thread WorkerThread_1: 333283335000
Thread WorkerThread_2: 333283335000範例 – Mutable 物件
Mutable 物件 (可變物件) 在創建後可以更改 (例如：Lists, Dictionaries, Sets)。 這與 Numbers, Strings, Tuples...等 immutable 物件形成對比，這些物件在建立後就無法更改。
當 cached 結果是 Mutable 物件時，您需要小心。
如果您修改它，更改將持續存在。 如果您沒有意識到，這可能會導致意外行為。如果您打算修改它，為了避免意外行為，您應該立即製作一個副本。
import copy
from functools import cached_property
class MutableDemo:
def __init__(self):
self._data = list(range(5))
@cached_property
def data(self):
return self._data
demo = MutableDemo()
# Make a copy
data_copy = demo.data[:] # Make a copy
data_copy.append(5)
print(demo.data) # Output: [0, 1, 2, 3, 4]
print(data_copy) # Output: [0, 1, 2, 3, 4, 5]Last updated
Was this helpful?