# functools 【可調用物件】 ## 簡介 *** functools module 用於簡化任務的工具箱，提供了一系列工具來以函數/方法式編程風格處理函數。 ## @cache *** `@functools.cache(user_function)` 用於向函數和方法新增記憶功能 ([Memoization](https://en.wikipedia.org/wiki/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。 {% hint style="info" %} **不同的 argument 模式 (類型) 被認為是不同的調用，將被單獨 cache。** 例如：`f(a=1, b=2)`, `f(b=2, a=1)`, 不同的 keyword argument 順序。 {% endhint %} {% hint style="info" %} **`@functools.cache` 只能與 Hashable 物件 一起使用(例如：Numbers, Strings, Tuples...)。** 因為 `@functools.cache` 將結果存儲在 Dictionary 中，並以參數作為 keys。 因此它不能直接與 unhashable 類型一起使用(例如：Lists, Dictionaries, Sets...)。 如果函數參數是可變的並且它們的值發生變化，則會擾亂 cache 機制。 {% endhint %} {% hint style="info" %} **通常最好只 cached pickled、 serialized 或純數據的物件類型，以避免潛在的問題。** Pickling 是將 Python 物件轉換為可以存儲在磁盤或透過網路發送的 byte stream。 並非所有 Python 物件都可以 pickle。 例如：打開的檔案或網路物件無法進行 pickle。 {% endhint %} ### 範例 – 費波那契數, 測量時間 *** `@cache` 包裝了 `fibonacci()` 遞迴函數，先前計算的結果將被 cache 和重用，從而顯著加快計算速度。 如果沒有 cache，遞迴函數將進行大量的重複計算。 首先測量執行時間。之後，Python 將檢查給定參數的函數結果是否在 cache 中。 如果是，Python 返回 cache 的結果； 如果不是，Python 計算後，將結果存儲在 cache 中。 {% hint style="warning" %} **在此範例中，`@functools.cache` 必須為第一個 decorator (因此最後寫入)。** 因為測量函數的實際執行時間更有意義，而不是返回 `@cache` 的結果。 {% endhint %} **時間複雜度表示法 (例如：`O(a*b)` 或 `O(2^n)`):** 為我們提供了一個高級的、概括的概念，即算法的運行時間如何隨著輸入大小的增加而增加。 但它並沒有直接告訴我們算法的絕對運行速度有多快，因為這還取決於算法中執行的特定 operations。 如範例，Fibonacci sequence 函數正在遞迴計算 35 個斐波那契數。 如果沒有記憶性，這將涉及大量的冗餘計算。 當函數被記憶時，它只需要計算每個 Fibonacci sequence 一次。因此，雖然 Fibonacci sequence 的原始時間複雜度確實是 `O(2^n)`，但記憶後的時間複雜度是 `O(n)`。 {% code title="PYTHON" %} ```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 ``` {% endcode %} ### 範例 – 昂貴操作 *** `expcious_operation()` 的結果將根據參數 `a` 和 `b` 進行 cache。 如果你使用相同的參數再次調用 `expcious_operation()`，Python 不會重新計算該函數，而是直接返回 cache 結果。 `expense_operation()` 函數運行兩個 nested loops， 這會導致執行一億次 iterations，每次 iteration 都涉及一次賦值操作。(記憶後的時間複雜度仍然維持 `O(a*b)`) {% code title="PYTHON" %} ```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 ``` {% endcode %} ### 範例 – 遞迴 vs 迴圈 *** {% code title="PYTHON" %} ```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 ``` {% endcode %} ## @lru\_cache *** 提供 LRU(Least Recently Used；最近最少使用) cache 的功能。 如果 cache 已滿，它將丟棄最近最少使用的項目。 {% code title="PYTHON" %} ```python lru_cache(maxsize=128, typed=False) ``` {% endcode %} * `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 類型差異 *** {% code title="PYTHON" %} ```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()) ``` {% endcode %} **執行結果：** ```TXT 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。 {% hint style="success" %} **在 multi-threaded 環境中使用 `@lru_cache` 時，確保 function operation, cache management 都是 thread-safe 的非常重要。** {% endhint %} {% hint style="info" %} **在計算 Fibonacci sequence 時，thread-safe 會變慢，原因是 lock 爭用。** 在 Fibonacci sequence 計算中，許多函數調用相同的參數。 `@lru_cache` 可以透過 cached 結果，來顯著加快這些調用的速度。 然而使用 lock，cache 調用必須等待 lock 可用。 在 thread 較多且調用重疊程度較高的場景中，lock 的等待時間可能會覆蓋 cache 的優勢。 {% endhint %} {% code title="PYTHON" %} ```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()) ``` {% endcode %} {% code title="PYTHON" %} ```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()) ``` {% endcode %} **Non-thread safe：** {% code title="PYTHON" %} ```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()) ``` {% endcode %} ## cache\_info() *** `cache_info()` 提供有關 cache 使用情況的報告(非 memory 的報告)。 包含： * `hits`：在 cache 中找到所需結果的次數。 表示有 `hits` 次不需要計算該函數，因為結果已經從之前的計算中獲得。 * `misses`：在 cache 中找不到所需結果，表示調用函數的次數。 * `maxsize`：cache 可以容納的最大大小，表示最多可以存儲 `maxsize` 個結果。 * `currsize`：當前 cache 的大小，表示已經存儲 `currsize` 個結果。 {% code title="PYTHON" %} ```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) ``` {% endcode %} ## cache\_clear() *** `cache_clear()` 用於手動清除所有 cache。 當您知道函數的結果可能會更改 (例如：結果取決於外部資源或狀態)，或者您想要釋放 memory 時，這非常有用。 {% hint style="info" %} **過於頻繁地使用 `cache_clear()` 可能會抵消 cache 優勢。如果底層數據或函數行為發生變化，不經常清除 cache 可能會導致使用過時的結果。** {% endhint %} {% code title="PYTHON" %} ```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)) ``` {% endcode %} **執行結果：** ``` [Simulating an expensive operation] 8 8 [Simulating an expensive operation] 27 [Simulating an expensive operation] 8 ``` ### 範例 – 外部 Data source 交互 *** 考慮一個與某些外部 data source 交互的函數(例如：數據庫、文件或 API)。 如果該 data source 發生更改，您需要確保沒有使用已過時的 cached 結果。 {% code title="" %} ```python 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 ``` {% endcode %} ## @cached\_property *** `@cached_property(func)` 用於 Class 中的方法新增記憶功能，並轉換為屬性(即，無需括號即可訪問， 計算出來的值與 Class 實例相關聯)。 該屬性的值在首次訪問時計算一次，然後在 Class 實例的 lifetime (生命週期) 內，作為一般屬性進行 cached。這使得 `@cached_property` 適合用於與實例相關的昂貴計算。 它的工作原理如下： * 首次訪問某個屬性時，Python 會檢查該名稱的屬性是否已存在。 * 如果不存在，則調用 `@cached_property` 方法，並將結果存儲為同名的屬性。 * 之後，cached 屬性的行為就像一般屬性一樣。 任何後續的讀取都將獲取 cached 值，任何寫入都將覆蓋 cached 值。 * 如果要清除 cached 值，應該使用 `del` 刪除該屬性。 下次訪問該屬性時，`@cached_property` 將再次運行。 {% hint style="success" %} **`@cached_property` 是針對每個實例的，而不是對每個 Clas 的。(Clas 的每個實例都將擁有自己的屬性 cached)** 因此如果您有多個 Clas 實例，它們可能會使用大量 memory。 {% endhint %} {% hint style="info" %} **`@property` 和 `@property.setter` 幫助控制屬性的設定和檢索方式，而 `@cached_property` 幫助 optimize 屬性的計算。** 當您需要控制或自定義屬性的設定或檢索方式時，請使用 `@property` 和 `@property.setter`。 {% endhint %} {% hint style="info" %} **`@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，並具有一些額外的靈活性，但代價是稍微複雜一些。 {% endhint %} ### Cached\_property vs. Cache *** **你應該兩者都使用嗎？** 在同一 Class 方法中，同時使用 `@cached_property`, `@cache` 通常是不必要的，並且可能會導致混亂。 這是因為它們是針對不同的場景而設計的：`@cached_property` 用於計算屬性值的方法，`@cache` 用於使用輸入，執行計算的函數。 然而，在某些情況下，使用兩者可能是有益的，例如：具有計算成本昂貴的 Class 方法，在內部調用另一個計算成本昂貴的函數。 總之，使用哪一種主要取決於您的具體用例、函數/方法的計算強度、函數/方法是否使用可變數據或有副作用、您環境的內存限制以及您的參數是否 函數/方法是可散列的。 {% hint style="success" %} **`@cached_property` 優點：** 當屬性計算量很大並且屬性被多次訪問時，它可以加速您的程序。 作為一個屬性，它不需要顯式調用函數或方法，從而使您的程式碼更清晰、更具可讀性。 目的用於建立唯讀屬性(請避免修改)。 cached 結果存儲在實例變數中。 {% endhint %} {% hint style="success" %} **`@cache` 優點：** 當函數計算量很大並且使用相同的參數多次調用該函數，它可以加速您的程序。 它可以 cached 具有任意數量參數的函數的結果。 {% endhint %} {% hint style="danger" %} **`@cached_property` 缺點：** 它只能與不帶參數的實例方法一起使用(`self` 除外)。 {% endhint %} {% hint style="danger" %} **`@cache` 缺點：** 它只能與接受 Hashable 參數的函數/方法一起使用，因為 cache 是根據 dictionary 實現的。 {% endhint %} {% hint style="danger" %} **共同缺點：** 如果函數有依賴於 mutable 數據，cache 可能容易導致不正確的結果。(將繼續返回舊的 cached 值) 它使用 memory 來存儲 cache，這在大數據或 memory 有限的環境中可能會出現問題。 {% endhint %} {% code title="PYTHON" %} ```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 ``` {% endcode %} ### 範例 – 基礎 *** {% code title="PYTHON" %} ```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 ``` {% endcode %} {% code title="PYTHON" %} ```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 ``` {% endcode %} ### 範例 – Thread-safe *** `@functools.cached_property` 是 thread-safe，這意味著它可以在 multithreaded 環境中安全使用。 如果沒有 thread-safe，一個 thread 可能會看到該屬性處於部分更新狀態，而另一個 thread 正在更新該屬性。 (看不到兩者之間的某些不一致狀態) {% code title="PYTHON" %} ```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) ``` {% endcode %} **執行結果：** ```TXT 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 物件形成對比，這些物件在建立後就無法更改。 {% hint style="success" %} **當 cached 結果是 Mutable 物件時，您需要小心。** 如果您修改它，更改將持續存在。 如果您沒有意識到，這可能會導致意外行為。如果您打算修改它，為了避免意外行為，您應該立即製作一個副本。 {% endhint %} {% code title="PYTHON" %} ```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] ``` {% endcode %} --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://docs.xiwind-corp.com/tech/python-library/functools-han-shu-gao-ji-cao-zuo.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.