> For the complete documentation index, see [llms.txt](https://docs.xiwind-corp.com/tech/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.xiwind-corp.com/tech/python-library/functools-han-shu-gao-ji-cao-zuo.md).

# 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，從而導致競爭條件，不可預測的結果。&#x20;

因此，在此類似的環境中使用 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 的報告)。&#x20;

包含：

* `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)**&#x20;

因此如果您有多個 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` 優點：**&#x20;

當屬性計算量很大並且屬性被多次訪問時，它可以加速您的程序。&#x20;

作為一個屬性，它不需要顯式調用函數或方法，從而使您的程式碼更清晰、更具可讀性。&#x20;

目的用於建立唯讀屬性(請避免修改)。 cached 結果存儲在實例變數中。
{% endhint %}

{% hint style="success" %}
**`@cache` 優點：**&#x20;

當函數計算量很大並且使用相同的參數多次調用該函數，它可以加速您的程序。&#x20;

它可以 cached 具有任意數量參數的函數的結果。
{% endhint %}

{% hint style="danger" %}
**`@cached_property` 缺點：**&#x20;

它只能與不帶參數的實例方法一起使用(`self` 除外)。
{% endhint %}

{% hint style="danger" %}
**`@cache` 缺點：**&#x20;

它只能與接受 Hashable 參數的函數/方法一起使用，因為 cache 是根據 dictionary 實現的。
{% endhint %}

{% hint style="danger" %}
**共同缺點：**&#x20;

如果函數有依賴於 mutable 數據，cache 可能容易導致不正確的結果。(將繼續返回舊的 cached 值)&#x20;

它使用 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 物件時，您需要小心。**&#x20;

如果您修改它，更改將持續存在。 如果您沒有意識到，這可能會導致意外行為。如果您打算修改它，為了避免意外行為，您應該立即製作一個副本。
{% 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 %}