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。

不同的 argument 模式 (類型) 被認為是不同的調用，將被單獨 cache。

例如：f(a=1, b=2), f(b=2, a=1), 不同的 keyword argument 順序。

@functools.cache 只能與 Hashable 物件 一起使用(例如：Numbers, Strings, Tuples...)。

因為 @functools.cache 將結果存儲在 Dictionary 中，並以參數作為 keys。 因此它不能直接與 unhashable 類型一起使用(例如：Lists, Dictionaries, Sets...)。 如果函數參數是可變的並且它們的值發生變化，則會擾亂 cache 機制。

通常最好只 cached pickled、 serialized 或純數據的物件類型，以避免潛在的問題。

Pickling 是將 Python 物件轉換為可以存儲在磁盤或透過網路發送的 byte stream。 並非所有 Python 物件都可以 pickle。 例如：打開的檔案或網路物件無法進行 pickle。

範例 – 費波那契數, 測量時間

---

@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)。

PYTHON

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

PYTHON

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 迴圈

---

PYTHON

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 已滿，它將丟棄最近最少使用的項目。

PYTHON

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 類型差異

---

PYTHON

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 的非常重要。

在計算 Fibonacci sequence 時，thread-safe 會變慢，原因是 lock 爭用。

在 Fibonacci sequence 計算中，許多函數調用相同的參數。 @lru_cache 可以透過 cached 結果，來顯著加快這些調用的速度。 然而使用 lock，cache 調用必須等待 lock 可用。 在 thread 較多且調用重疊程度較高的場景中，lock 的等待時間可能會覆蓋 cache 的優勢。

PYTHON

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

PYTHON

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：

PYTHON

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 個結果。

PYTHON

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 時，這非常有用。

過於頻繁地使用 cache_clear() 可能會抵消 cache 優勢。如果底層數據或函數行為發生變化，不經常清除 cache 可能會導致使用過時的結果。

PYTHON

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。

@property 和 @property.setter 幫助控制屬性的設定和檢索方式，而 @cached_property 幫助 optimize 屬性的計算。

當您需要控制或自定義屬性的設定或檢索方式時，請使用 @property 和 @property.setter。

@cached_property 要求每個實例上的 __dict__ 是 mutable mapping。 這意味著不適用於 Metaclasses，因為 Metaclasses 的 __dict__ 是 Class 名稱空間的 read-only proxies。

@cached_property 可能會干擾 Key-sharing dictionary (PEP 412) 的操作，導致實例 dictionary 中的空間使用量增加。 如果優先考慮節省空間的 Key-sharing，或者如果 mutable mapping 不可用，則可以通過在 @lru_cache() 之上使用 property() 來實現與 @cached_property 類似的效果。 這種組合將為方法提供 cache，並具有一些額外的靈活性，但代價是稍微複雜一些。

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 有限的環境中可能會出現問題。

PYTHON

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

範例 – 基礎

---

PYTHON

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

PYTHON

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 正在更新該屬性。 (看不到兩者之間的某些不一致狀態)

PYTHON

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 物件時，您需要小心。

如果您修改它，更改將持續存在。 如果您沒有意識到，這可能會導致意外行為。如果您打算修改它，為了避免意外行為，您應該立即製作一個副本。

PYTHON

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]