from functools import partial

def power(base, exp, *, mod=None):
    res = base ** exp
    return res if mod is None else res % mod

# 1.1 預(yù)綁定部分位置參數(shù)
square = partial(power, exp=2)      # 固定指數(shù)
print(square(3))                    # 9
print(square(3, mod=5))             # 4

# 1.2 預(yù)綁定關(guān)鍵字參數(shù)
cube_mod_7 = partial(power, exp=3, mod=7)
print(cube_mod_7(2))                # 1  (8 % 7)

# 1.3 后續(xù)調(diào)用的關(guān)鍵字**可以覆蓋**先前綁定的關(guān)鍵字
p = partial(power, exp=2, mod=5)
print(p(3))                         # 4   (9 % 5)
print(p(3, mod=None))               # 9   —— 覆蓋為 None

# 1.4 后續(xù)調(diào)用的**位置參數(shù)不能“挪位”覆蓋**已綁定的位置參數(shù)
mul = lambda a, b, c: (a, b, c)
p2 = partial(mul, 10)               # a=10 已固定
print(p2(20, 30))                   # (10, 20, 30)
# p2(5, a=1) -> TypeError: a 給了多個(gè)值（不允許）

from functools import partial, reduce
from operator import mul

nums = [1, 2, 3, 4]

# 2.1 map：把二元函數(shù)“變成一元”
double = partial(mul, 2)            # 固定左操作數(shù)
print(list(map(double, nums)))      # [2, 4, 6, 8]

# 2.2 sorted：固定 key / reverse
students = [{"name":"A", "age":20}, {"name":"B", "age":18}]
by = partial(sorted, key=lambda x: x["age"])
print(by(students))                 # 按 age 升序
by_desc = partial(sorted, key=lambda x: x["age"], reverse=True)
print(by_desc(students))

# 2.3 reduce：固定初始值
sum_from_10 = partial(reduce, lambda a, b: a + b, initial=10)
print(sum_from_10(nums))            # 20

from functools import partial
import asyncio, concurrent.futures, time

def on_done(label, fut: concurrent.futures.Future):
    print(f"[{label}] result ->", fut.result())

def heavy(x):
    time.sleep(0.2)
    return x * x

async def main():
    loop = asyncio.get_running_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        fut = loop.run_in_executor(pool, heavy, 9)
        # 給回調(diào)多傳一個(gè) label
        fut.add_done_callback(partial(on_done, "task#1"))
        # 也可以 await 等它
        print("await:", await fut)

asyncio.run(main())

from functools import wraps, partial

def _retry_impl(func, attempts, delay):
    @wraps(func)
    def wrapper(*args, **kwargs):
        last = None
        for _ in range(attempts):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                last = e
                time.sleep(delay)
        raise last
    return wrapper

# 用 partial 固定 attempts/delay，得到一個(gè)“可當(dāng)裝飾器用”的可調(diào)用
import time
retry3 = partial(_retry_impl, attempts=3, delay=0.1)

@retry3
def flaky():
    print("try...")
    if time.time() % 2 < 1:
        raise ValueError("boom")
    return "ok"

print("flaky ->", flaky())

from functools import partial
def f(a, b, c): return (a, b, c)

# partial 只能“從左到右”補(bǔ)位置參數(shù)（或直接用關(guān)鍵字）
g1 = partial(f, 1)                 # ==> f(1, b, c)

# lambda 可自由重排
g2 = lambda b, c: f(1, c, b)       # 調(diào)換了 b/c 的位置
print(g1(2, 3), g2(2, 3))

from functools import partial

def greet(greet_word, name, punctuation="!"):
    return f"{greet_word}, {name}{punctuation}"

hi_tom = partial(greet, "Hi", "Tom", punctuation=".")
print(hi_tom())                    # Hi, Tom.

print(hi_tom.func)                 # 原函數(shù) <function greet ...>
print(hi_tom.args)                 # ('Hi', 'Tom')
print(hi_tom.keywords)             # {'punctuation': '.'}

# 可覆蓋同名關(guān)鍵字
print(hi_tom(punctuation="!!!"))   # Hi, Tom!!!

from functools import partialmethod

class Logger:
    def log(self, level, msg):
        print(f"[{level}] {self.name}: {msg}")

    debug = partialmethod(log, "DEBUG")   # 等價(jià)于 def debug(self, msg): return self.log("DEBUG", msg)
    info  = partialmethod(log, "INFO")
    def __init__(self, name):
        self.name = name

l = Logger("core")
l.debug("hello")   # [DEBUG] core: hello
l.info("world")    # [INFO] core: world

from functools import partial, update_wrapper

def power(a, b): return a ** b

square = partial(power, b=2)

def as_func(p):
    def wrapper(*args, **kwargs):
        return p(*args, **kwargs)
    return update_wrapper(wrapper, p.func)

square_fn = as_func(square)
print(square_fn.__name__, square_fn.__doc__)   # 繼承了 power 的元數(shù)據(jù)

from functools import partial
from concurrent.futures import ProcessPoolExecutor, as_completed

def area(w, h, scale=1.0):
    return w * h * scale

if __name__ == "__main__":
    scaled_area = partial(area, scale=0.5)  # 可被 pickle
    items = [(10, 20), (3, 4), (6, 7)]
    with ProcessPoolExecutor() as ex:
        futs = [ex.submit(scaled_area, w, h) for (w, h) in items]
        for f in as_completed(futs):
            print("area:", f.result())

from functools import partial

print_info = partial(print, "[INFO]")      # 固定前綴
print_warn = partial(print, "[WARN]")
print_info("system started")
print_warn("disk almost full")