Python 深入（附 Go 速覽）

CPython 執行模型：從 .py 到 bytecode 到 PVM 進階

「Python 是直譯語言」這句話只對一半。我們平常用的 CPython（用 C 寫的官方實作）其實會先把原始碼編譯成 bytecode，再交給一個 stack-based 的虛擬機器（Python Virtual Machine，PVM，也就是 ceval.c 裡那個巨大的 eval loop）一條一條執行。

編譯流程與 .pyc

1. Tokenize → Parse：3.9 起改用 PEG parser，產生 AST（可用 ast.parse() 看）。
2. Compile：建立符號表（決定每個名字是 local、global、free 還是 cell variable），產生 code object（func.__code__），裡面有 co_code（bytecode）、co_consts、co_varnames 等。
3. 快取：被 import 的模組會把 code object 序列化（marshal）成 __pycache__/mod.cpython-312.pyc，下次若原始碼 mtime 沒變就直接載入，省去編譯。直接執行的主程式（python main.py）不會寫 .pyc。
4. 執行：PVM 為每次函式呼叫建立一個 frame（含 local 變數陣列、value stack、指令指標），在 eval loop 中 dispatch 每個 opcode。

用 dis 看 bytecode

import dis

def add(a, b):
    c = a + b
    return c

dis.dis(add)

  # Python 3.14 的輸出（不同版本 opcode 會不同）
  3           RESUME                   0

  4           LOAD_FAST_BORROW_LOAD_FAST_BORROW 1 (a, b)
              BINARY_OP                0 (+)
              STORE_FAST               2 (c)

  5           LOAD_FAST_BORROW         2 (c)
              RETURN_VALUE

讀法：左邊是原始碼行號，中間是 opcode，右邊是參數與註解。PVM 是 stack machine：LOAD_FAST 把 local 變數推上 value stack，BINARY_OP (+) 彈出兩個、呼叫 a.__add__(b)（失敗再試 b.__radd__(a)），結果推回去。在 3.11 以前的版本你會看到 LOAD_FAST a、LOAD_FAST b 分開兩條；3.14 把它們融合成一條 superinstruction。

3.11+ 的 specializing adaptive interpreter 與 JIT

Python 3.11（PEP 659，Faster CPython 計畫）引入自適應特化：一條指令被執行多次後，會依實際觀察到的型別替換成特化版本，例如 BINARY_OP 在兩邊都是 int 時變成 BINARY_OP_ADD_INT，LOAD_ATTR 變成直接用 offset 讀 instance 屬性的版本。若型別改變則「去特化」退回通用版。這讓 3.11 平均比 3.10 快約 1.25 倍。可用 dis.dis(f, adaptive=True) 看特化後的指令。

3.13 起有實驗性的 copy-and-patch JIT（PEP 744，需編譯時開啟），目前效益仍有限。重點觀念：CPython 仍然是以直譯為主，每個 opcode 都有 dispatch 成本、每個物件操作都要走動態查找，這是純 Python 迴圈比 C 慢 10–100 倍的根本原因。

物件模型：一切皆物件、identity 與參數傳遞 高頻

在 CPython 裡，每個值都是一個 heap 上的 C 結構 PyObject，至少包含兩個欄位：reference count（ob_refcnt）與型別指標（ob_type）。整數、函式、類別、模組、甚至 None 都是物件。變數不是「裝值的盒子」，而是貼在物件上的名字標籤（name binding）。

x = 42
print(type(x), type(type(x)), type(len))   # <class 'int'> <class 'type'> <class 'builtin_function_or_method'>
print(isinstance(type, object), isinstance(object, type))  # True True（type 與 object 互相纏繞）

def f(): pass
f.custom_attr = "函式也是物件，可以掛屬性"
print(f.__code__.co_name, f.custom_attr)

identity（is / id）vs equality（==）

id(obj)

物件在其生命週期內唯一的整數；CPython 中就是記憶體位址。物件被回收後 id 可能被重用。

a is b

比較 identity：是不是同一個物件。只是指標比較，不能被 overload，非常快。

a == b

比較 value：呼叫 a.__eq__(b)，可以自訂。預設（object 的實作）退化成 identity。

規則：跟 singleton（None、True/False、Ellipsis、sentinel 物件）比較用 is，其餘都用 ==。PEP 8 也這樣要求。用 == None 的風險是某些物件（例如 NumPy array）會 overload __eq__ 回傳非 bool。

small int cache 與字串 interning

a = 256; b = 256
print(a is b)          # True：CPython 預先建立 -5 ~ 256 的整數並重複使用

x = int("257"); y = int("257")
print(x is y)          # False：執行期產生的兩個不同物件
print(x == y)          # True

# 注意：同一個 code object 裡的相同常數會被合併，所以下面在 .py 檔中可能是 True
p = 257; q = 257
print(p is q)          # 在同一個模組/函式中編譯 → 常數去重 → 通常 True；REPL 分兩行輸入則 False

s1 = "hello_world"; s2 = "hello" + "_world"
print(s1 is s2)        # True：看起來像識別字的字串常數會被 intern，且 "hello"+"_world" 在編譯期常數摺疊

import sys
t1 = "".join(["hel", "lo world!"]); t2 = "hello world!"
print(t1 is t2, sys.intern(t1) is sys.intern(t2))   # False True

mutable vs immutable

「不可變」指物件本身內容不能改；x += 1 對 int 來說是建立新物件再把名字 x 重新綁定。注意 tuple 不可變但可以裝可變物件：t = ([],) 中的 list 仍可 append，這樣的 tuple 也就不可 hash。

t = ([1, 2],)
t[0].append(3)        # OK，t == ([1, 2, 3],)
try:
    t[0] += [4]       # 經典謎題：會丟 TypeError，但 list 其實已經被改了！
except TypeError as e:
    print(e)          # 'tuple' object does not support item assignment
print(t)              # ([1, 2, 3, 4],)
# 原因：t[0] += [4] 展開成 tmp = t[0]; tmp = tmp.__iadd__([4])（就地修改成功）; t[0] = tmp（失敗）

參數傳遞：pass-by-object-reference（call by sharing）

Python 既不是 pass-by-value 也不是 C++ 意義上的 pass-by-reference。函式呼叫時，參數名字被綁定到呼叫端傳入的同一個物件。因此：

• 在函式內就地修改（mutate）可變物件 → 呼叫端看得到。
• 在函式內重新綁定（rebind）參數名字 → 只影響 local 名字，呼叫端看不到。

def mutate(lst):
    lst.append(99)      # 改的是同一個物件

def rebind(lst):
    lst = [0]           # 只是讓 local 名字指向新物件

nums = [1, 2]
mutate(nums); print(nums)   # [1, 2, 99]
rebind(nums); print(nums)   # [1, 2, 99]（沒變）

def inc(n):
    n += 1              # int 不可變，+= 產生新物件並 rebind local 名字
    return n
k = 1; inc(k); print(k)     # 1

記憶體管理：reference counting、generational GC 與省記憶體技巧 深入

Reference counting：主要的回收機制

每個物件有 ob_refcnt。名字綁定、放進容器、作為參數傳遞都會 Py_INCREF；名字離開 scope、del、從容器移除會 Py_DECREF。計數歸零的那一刻物件立即被釋放（呼叫 tp_dealloc），所以 CPython 的記憶體回收大部分是確定性的，這也是為什麼 open() 忘了 close 在 CPython 上通常「剛好沒事」（但在 PyPy 上不行，所以仍要用 with）。

import sys
a = []
print(sys.getrefcount(a))   # 2：a 本身 + 傳進 getrefcount 的暫時參考
b = a
print(sys.getrefcount(a))   # 3
del b
print(sys.getrefcount(a))   # 2

優點：即時回收、記憶體峰值低、行為可預期。缺點：(1) 每次賦值都要改計數，有開銷；(2) 多執行緒同時改計數會 race，這正是 GIL 存在的主因之一；(3) 無法處理循環參考。

循環參考與 generational GC

import gc

class Node:
    def __init__(self, name):
        self.name = name
        self.peer = None

a, b = Node("a"), Node("b")
a.peer, b.peer = b, a      # a → b → a 形成環
del a, b                    # refcount 都還是 1，refcounting 回收不了
print(gc.collect())         # 回傳回收的不可達物件數（> 0）

CPython 另有一個 cycle detector（gc 模組），只追蹤「可能形成環的容器物件」（list、dict、自訂 instance 等；int、str 不追蹤）。演算法概念：對候選集合中每個物件，把「來自集合內部的參考」從 refcount 中扣掉，剩下仍 > 0 的代表有外部參考（可達），從它們出發標記可達物件，其餘就是只被環內互相參考的垃圾。

分代（generational）假設「大部分物件很年輕就死」：新物件放在第 0 代，活過一次收集就晉升。傳統 CPython（≤ 3.13）有三代，預設 threshold 為 (700, 10, 10)：第 0 代「配置數 − 釋放數」超過 700 觸發 gen0 收集，gen0 收集 10 次觸發一次 gen1，以此類推。3.14 改為 incremental GC（young / old 兩代，每次只掃描 old 的一部分以縮短停頓），在 3.14 上 gc.get_threshold() 會回傳 (2000, 10, 0)。

__del__ 的陷阱

• __del__ 是 finalizer，不是 destructor 保證：何時被呼叫不確定（有環時要等 GC；PyPy 上更晚），直譯器結束時甚至可能不呼叫。
• __del__ 內丟出的例外會被忽略（只印警告到 stderr）。
• 在 __del__ 中把 self 存到 global 會「復活」物件（resurrection）。
• 直譯器關閉時，模組的 global 可能已被設成 None，__del__ 裡用到 global 會炸。
• 3.4 之前，環中有 __del__ 的物件無法回收（放進 gc.garbage）；PEP 442 之後可以回收，但呼叫順序不保證。

結論：釋放資源請用 context manager（with）或 weakref.finalize，不要依賴 __del__。

weakref：不增加 refcount 的參考

import weakref

class BigModel:
    def __init__(self, name): self.name = name

cache = weakref.WeakValueDictionary()   # value 被回收後自動從 dict 消失
m = BigModel("ctr-v3")
cache["ctr"] = m
print(cache.get("ctr").name)   # ctr-v3
del m                          # 最後一個強參考消失
print(cache.get("ctr"))        # None（CPython 上立即生效）

# 樹狀結構：child 用 weakref 指向 parent，避免 parent ↔ child 循環
class TreeNode:
    def __init__(self, parent=None):
        self._parent = weakref.ref(parent) if parent else None
        self.children = []
    @property
    def parent(self):
        return self._parent() if self._parent else None

用途：快取（不想讓快取延長物件壽命）、observer/callback 註冊表、打破 parent-child 環。注意 int、str、tuple、list、dict 等內建型別不能直接被 weakref（但它們的子類別可以）。

sys.getsizeof 與 __slots__

sys.getsizeof(obj) 只回傳物件本身的大小（shallow），不含它參考的物件。例如 sys.getsizeof([big_str] * 1000) 只算 list 的指標陣列。要算深度大小需自己遞迴、或用 tracemalloc / pympler。

import sys, tracemalloc

class P:                       # 一般 class：屬性存在 instance 的 __dict__
    def __init__(self, x, y): self.x, self.y = x, y

class PS:                      # __slots__：屬性存在固定 offset 的 descriptor 槽位
    __slots__ = ("x", "y")
    def __init__(self, x, y): self.x, self.y = x, y

for cls in (P, PS):
    tracemalloc.start()
    objs = [cls(i, i) for i in range(100_000)]
    current, _ = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    print(cls.__name__, f"{current / 1e6:.1f} MB")
# 在 3.14 上實測約 P 9.6 MB vs PS 5.6 MB（含 list 本身與 int 物件）

ps = PS(1, 2)
# ps.z = 3   # AttributeError：沒有 __dict__，不能動態加屬性

__slots__ 的取捨：省記憶體、屬性存取稍快；但不能動態加屬性、預設不能被 weakref（要把 "__weakref__" 加進 slots）、多重繼承時多個有非空 slots 的父類別會衝突、子類別若沒宣告 __slots__ 又會長回 __dict__。適合「數百萬個小物件」的場景，例如在記憶體中保存大量事件或特徵。dataclass 可用 @dataclass(slots=True)（3.10+）。

GIL：是什麼、為什麼存在、怎麼繞過 高頻

Global Interpreter Lock 是 CPython 直譯器層級的一把 mutex：同一時間只有一個 OS thread 能執行 Python bytecode。多個 thread 仍然是真的 OS thread（由 kernel 排程），但要執行 bytecode 前必須先拿到 GIL。

為什麼要有 GIL

• 保護 reference count：ob_refcnt++ 不是 atomic，若兩個 thread 同時改，計數錯了就會 use-after-free 或記憶體洩漏。若對每個物件加鎖或用 atomic 操作，單執行緒效能會明顯下降。
• 保護直譯器內部狀態：記憶體配置器（pymalloc）、內建型別的內部結構（dict、list 的 resize）等。
• 讓 C 擴充好寫：C 擴充作者不必處理細粒度鎖，這是 Python 早期生態（NumPy 等）快速成長的重要原因。
• 歷史上（1990 年代）多核心機器罕見，一把大鎖是最簡單、單執行緒最快的設計。

GIL 何時釋放

• 阻塞 I/O 時：socket.recv、file.read、time.sleep 等在進入系統呼叫前釋放 GIL，回來後再搶。
• 時間片到期：3.2 起的「new GIL」，持有者執行超過 sys.getswitchinterval()（預設 0.005 秒 = 5ms），且有其他 thread 在等，就會被要求釋放。
• C 擴充主動釋放：NumPy 的大型運算、hashlib 對大資料計算、zlib 壓縮等用 Py_BEGIN_ALLOW_THREADS 釋放 GIL，這段期間可以真正平行。

對 CPU-bound vs IO-bound 的影響

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def cpu_task(_=None, n=5_000_000):   # 第一個參數吃掉 map 傳入的 index
    s = 0
    for i in range(n):
        s += i * i
    return s

def io_task(_=None):
    time.sleep(0.5)      # 模擬網路 I/O，sleep 期間釋放 GIL

def bench(executor_cls, fn, workers=4):
    t = time.perf_counter()
    with executor_cls(max_workers=workers) as ex:
        list(ex.map(fn, range(workers)))
    return time.perf_counter() - t

if __name__ == "__main__":        # ProcessPool 在 spawn 模式下必須有這個 guard
    print("CPU  threads :", bench(ThreadPoolExecutor, cpu_task))   # ≈ 單執行緒 × 4（甚至更慢）
    print("CPU  procs   :", bench(ProcessPoolExecutor, cpu_task))  # ≈ 單執行緒 × 1（4 核心平行）
    print("I/O  threads :", bench(ThreadPoolExecutor, io_task))    # ≈ 0.5 秒，I/O 期間可重疊

結論：IO-bound 用 threading 或 asyncio 很有效；CPU-bound 的純 Python 程式用 threading 無法加速，要用 multiprocessing、把運算下放到會釋放 GIL 的 C 擴充（NumPy），或用 free-threaded build。

import threading

counter = 0
lock = threading.Lock()

def unsafe():
    global counter
    for _ in range(100_000):
        counter += 1           # 非 atomic：LOAD_GLOBAL, BINARY_OP, STORE_GLOBAL

def safe():
    global counter
    for _ in range(100_000):
        with lock:
            counter += 1

threads = [threading.Thread(target=unsafe) for _ in range(8)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)   # 可能小於 800000（是否重現視版本與時機而定）

Free-threaded Python（PEP 703）現況

• 3.13（2024/10）：提供實驗性的 free-threaded build（執行檔通常叫 python3.13t），可在編譯時移除 GIL。用 sys._is_gil_enabled() 檢查。技術上改用 biased reference counting（擁有者 thread 用非 atomic 計數、其他 thread 用 atomic）、immortal objects（PEP 683，None、小整數等 refcount 不變）、per-object lock 保護 dict/list、改用 mimalloc 配置器。單執行緒效能有明顯折損，且 specializing interpreter 在該模式下被停用。
• 3.14（2025/10）：PEP 779 讓 free-threaded build 進入「officially supported」階段（不再標示 experimental），但仍不是預設；specialization 已在 free-threaded 模式恢復，單執行緒折損大幅縮小。
• 生態：C 擴充必須宣告支援（Py_mod_gil）並重新編譯，否則載入時會重新啟用 GIL。NumPy、Cython 等主流套件陸續支援中。

內建資料結構內部：dict、set、list 與時間複雜度 高頻

dict：open addressing 的 hash table

CPython 的 dict 是 hash table，碰撞處理用 open addressing（不是 Java HashMap 的 separate chaining）：計算 hash(key) 取低位元得到 slot，若已被佔用就依照「j = (5*j + 1 + perturb) mod 2^k，perturb >>= 5」的公式跳到下一個位置探測，逐步把 hash 的高位元也納入，避免聚集。

3.6 起是 compact dict（Raymond Hettinger 提出）：拆成兩個陣列——

• indices：稀疏的小整數陣列（依 table 大小用 int8/16/32/64），存 entries 的 index 或 EMPTY/DUMMY。
• entries：密集陣列，依插入順序存 (hash, key, value)。

d = {"timmy": "red", "barry": "green", "guido": "blue"}

indices = [None, 1, None, None, None, 0, None, 2]     # 稀疏，每格只要 1 byte
entries = [[-9092791511155847987, 'timmy', 'red'],     # 密集、按插入順序
           [-8522787127447073495, 'barry', 'green'],
           [-6480567542315338377, 'guido', 'blue']]

好處：(1) 記憶體減少 20–25%（稀疏的部分只存小整數）；(2) 迭代只需掃密集的 entries，更快；(3) 天然保持插入順序。3.6 是實作細節，3.7 起語言規範保證 dict 保持插入順序。刪除時 entries 留下空洞（DUMMY），resize 時壓縮。

當使用量達到容量的約 2/3 時 resize（重新配置更大的 table 並重新插入），所以 insert 是 amortized O(1)。最壞情況（大量碰撞）退化到 O(n)。

hash / eq 契約

• 若 a == b 則必須 hash(a) == hash(b)（反之不必成立，hash 相同只是碰撞）。
• 物件作為 key 期間，hash 值不可改變。
• 查找流程：先比 hash 值（快速排除），再比 identity（is），最後才呼叫 __eq__。
• 自訂 __eq__ 卻沒定義 __hash__ 時，Python 會把 __hash__ 設為 None，該類別變成 unhashable。

class Point:
    def __init__(self, x, y): self.x, self.y = x, y
    def __eq__(self, other):
        return isinstance(other, Point) and (self.x, self.y) == (other.x, other.y)
    def __hash__(self):
        return hash((self.x, self.y))       # 用參與 __eq__ 的同一組欄位

print({Point(1, 2): "ok"}[Point(1, 2)])     # ok

# 違反契約的後果：key 放進去之後被改，就再也找不到
p = Point(1, 2); d = {p: "v"}
p.x = 100
print(p in d)   # False：用新的 hash 去找 slot，找不到；舊 slot 裡的物件又不相等
# 正確做法：key 物件應該不可變，例如 @dataclass(frozen=True) 或 NamedTuple

print(hash(1) == hash(1.0) == hash(True))   # True，且 1 == 1.0 == True，所以 {1: 'a', True: 'b'} 結果是 {1: 'b'}

為什麼 list 不能當 dict key？

因為 list 可變。若允許用內容計算 hash，list 被修改後 hash 就變了，dict 會到錯誤的 slot 找它（上面 Point 的例子）。若改用 identity 計算 hash，那 d[[1,2]] 永遠找不到「內容相同的另一個 list」，失去意義。所以 Python 選擇讓 list 的 __hash__ = None。需要時轉成 tuple(lst)（前提是元素都 hashable）或 frozenset。

set

set 也是 open addressing hash table，但只存 key（沒有 value）、不保證順序（沒有 compact 化，探測策略是先線性探測數格再跳躍，對 cache 較友善）。適合成員檢查（x in s O(1)）與集合運算。frozenset 是不可變版本，可以當 dict key。

list：動態陣列（over-allocation）

list 底層是 PyObject* 指標的連續陣列（存的是指標，不是物件本身）。append 時若容量不足就 realloc 到更大的大小，成長因子約 1.125 倍加上常數（new = n + (n >> 3) + 6，再對齊到 4 的倍數），因此 append 是 amortized O(1)。

import sys
lst, prev = [], sys.getsizeof([])
for i in range(40):
    lst.append(i)
    if (size := sys.getsizeof(lst)) != prev:
        print(f"len={len(lst):2d} bytes={size}")
        prev = size
# 3.14 實測：len=1 → 88, 5 → 120, 9 → 184, 17 → 248, 25 → 312, 33 → 376
# 每次擴容不是每次 append，這就是 amortized O(1)

時間複雜度總表

from collections import deque, Counter, defaultdict, OrderedDict
import heapq

# 1) BFS 一定用 deque，list.pop(0) 是 O(n)
start = "A"
q = deque([start]); seen = {start}

# 2) Top-K：100 萬個廣告的 CTR 取前 10 名，O(n log k)
scores = {"ad1": 0.031, "ad2": 0.054, "ad3": 0.012, "ad4": 0.047}
print(heapq.nlargest(2, scores.items(), key=lambda kv: kv[1]))   # [('ad2', 0.054), ('ad4', 0.047)]

# 3) Counter：Wordle 判題（處理重複字母的經典寫法）
def wordle(answer: str, guess: str) -> str:
    res = ["B"] * len(guess)                 # B=灰, Y=黃, G=綠
    remain = Counter()
    for i, (a, g) in enumerate(zip(answer, guess)):
        if a == g:
            res[i] = "G"
        else:
            remain[a] += 1                   # 只有沒對上綠色的字母才能拿來配黃色
    for i, g in enumerate(guess):
        if res[i] != "G" and remain[g] > 0:
            res[i] = "Y"
            remain[g] -= 1
    return "".join(res)
print(wordle("apple", "paper"))  # YYGYB

# 4) LRU cache 用 OrderedDict（move_to_end + popitem(last=False) 都是 O(1)）
class LRU:
    def __init__(self, cap): self.cap, self.d = cap, OrderedDict()
    def get(self, k):
        if k not in self.d: return None
        self.d.move_to_end(k); return self.d[k]
    def put(self, k, v):
        self.d[k] = v; self.d.move_to_end(k)
        if len(self.d) > self.cap: self.d.popitem(last=False)

Appier 真題：合併兩個 List[dict] 中相同 id 的資料

from collections import defaultdict
from typing import Any

def merge_by_id(*sources: list[dict[str, Any]], key: str = "id") -> list[dict[str, Any]]:
    """O(N) 合併多個來源的 user attributes；後面來源覆蓋前面，保留第一次出現的順序。"""
    merged: dict[Any, dict[str, Any]] = {}       # dict 保持插入順序（3.7+ 保證）
    for src in sources:
        for row in src:
            merged.setdefault(row[key], {}).update(row)
    return list(merged.values())

a = [{"id": 1, "name": "Amy"}, {"id": 2, "name": "Bob"}]
b = [{"id": 2, "age": 30}, {"id": 3, "name": "Cat", "age": 22}]
print(merge_by_id(a, b))
# [{'id': 1, 'name': 'Amy'}, {'id': 2, 'name': 'Bob', 'age': 30}, {'id': 3, 'name': 'Cat', 'age': 22}]

常見陷阱：每一個都可能是面試題 高頻

1. Mutable default arguments

def add_tag(tag, tags=[]):        # 預設值在 def 執行時「只求值一次」
    tags.append(tag)
    return tags

print(add_tag("a"))   # ['a']
print(add_tag("b"))   # ['a', 'b']  ← 同一個 list 被所有呼叫共用
print(add_tag.__defaults__)        # (['a', 'b'],) 預設值就存在函式物件上

def add_tag_fixed(tag, tags=None):
    if tags is None:
        tags = []
    tags.append(tag)
    return tags

為什麼會這樣：def 是一個在執行期執行的陳述句，它建立函式物件時把預設值算好存進 __defaults__。之後每次呼叫若沒傳參數，就用同一個物件。這個設計讓預設值計算只發生一次（效能），也讓「用預設參數當函式層級快取」成為可能（但不推薦）。

2. Late binding closures（迴圈中的 lambda）

funcs = [lambda: i for i in range(3)]
print([f() for f in funcs])        # [2, 2, 2]，不是 [0, 1, 2]

# 原因：closure 捕捉的是「變數 i」（cell），不是當下的值；呼叫時才查 i，此時 i 已經是 2
funcs = [lambda i=i: i for i in range(3)]          # 解法 1：用預設參數在定義時綁值
print([f() for f in funcs])        # [0, 1, 2]

from functools import partial
funcs = [partial(lambda x: x, i) for i in range(3)] # 解法 2：partial
print([f() for f in funcs])        # [0, 1, 2]

同樣的問題出現在 for 迴圈裡註冊 callback、建立 thread（Thread(target=lambda: work(i))）時。Go 在 1.22 之前也有一模一樣的問題（goroutine 捕捉迴圈變數），1.22 起改成每次迭代一個新變數。

3. is vs ==

x = 1000
print(x is 1000)     # SyntaxWarning: "is" with 'int' literal；結果依實作而定
print([] is [])      # False：兩個不同物件
print(None == None, None is None)   # 都 True，但請寫 is None

4. 迭代中修改容器

nums = [1, 2, 2, 3, 4]
for n in nums:
    if n % 2 == 0:
        nums.remove(n)       # 刪除後元素左移，iterator 的 index 照常 +1，會跳過元素
print(nums)                  # [1, 2, 3]  ← 第二個 2 被跳過了

nums = [1, 2, 2, 3, 4]
nums = [n for n in nums if n % 2]       # 正確：建立新 list
print(nums)                  # [1, 3]

d = {"a": 1, "b": 2}
# for k in d: del d[k]       # RuntimeError: dictionary changed size during iteration
for k in list(d):            # 先複製 keys
    if d[k] > 1: del d[k]

5. 浮點數

print(0.1 + 0.2)                 # 0.30000000000000004（IEEE 754 二進位無法精確表示 0.1）
print(0.1 + 0.2 == 0.3)          # False
import math
print(math.isclose(0.1 + 0.2, 0.3))      # True：比較浮點數用容差

from decimal import Decimal
print(Decimal("0.1") + Decimal("0.2"))   # 0.3：金額計算用 Decimal（注意要傳字串）
print(round(2.5), round(3.5))    # 2 4：banker's rounding（四捨六入五成雙）
print(int(-3.7), -7 // 2, -7 % 2) # -3 -4 1：int() 往 0 截斷；// 往負無限大取 floor

6. 淺拷貝 vs 深拷貝

import copy
grid = [[0] * 3] * 3          # 陷阱：外層 list 裡是「同一個」內層 list 的 3 個參考
grid[0][0] = 1
print(grid)                   # [[1, 0, 0], [1, 0, 0], [1, 0, 0]]
grid = [[0] * 3 for _ in range(3)]   # 正確

orig = {"tags": ["a"], "n": 1}
shallow = orig.copy()         # 或 dict(orig)、copy.copy(orig)、list 用 lst[:]
deep = copy.deepcopy(orig)    # 遞迴複製（會處理循環參考，透過 memo dict）
orig["tags"].append("b")
print(shallow["tags"], deep["tags"])   # ['a', 'b'] ['a']

淺拷貝只複製最外層容器，內部元素仍是同一批物件；深拷貝遞迴複製，成本高（且對大型結構很慢）。在 API 服務中，若把全域設定 dict 回傳給呼叫端並被修改，就會汙染所有後續請求——回傳 MappingProxyType（唯讀 view）或 frozen dataclass 更安全。

7. 其他經典

• UnboundLocalError：函式內對變數賦值，編譯器就把它當 local；賦值前讀取會炸。要改外層變數用 global / nonlocal。
• class 屬性被所有 instance 共用：class A: items = []，所有 instance 的 self.items.append 都改到同一個 list。
• except: 裸捕捉：會連 KeyboardInterrupt、SystemExit 一起吞掉。
• 字串累加：迴圈中 s += x 最壞 O(n²)（CPython 有時能就地優化但不保證），用 "".join(parts)。
• generator 只能迭代一次：第二次迭代是空的，不會報錯。
• circular import：模組互相 import 時，取用到尚未執行完的模組屬性會 ImportError/AttributeError。

Iterator、Generator 與 itertools 進階

迭代協定

• Iterable：有 __iter__()，回傳 iterator（list、dict、str、file）。
• Iterator：有 __next__()（沒東西時 raise StopIteration）且 __iter__() 回傳自己。
• for x in obj 等價於：it = iter(obj)，反覆 next(it) 直到 StopIteration。

class Countdown:                 # iterable：每次 iter() 都回傳新的 iterator，可重複迭代
    def __init__(self, n): self.n = n
    def __iter__(self):
        n = self.n
        while n > 0:
            yield n              # __iter__ 寫成 generator 是最簡潔的做法
            n -= 1

c = Countdown(3)
print(list(c), list(c))          # [3, 2, 1] [3, 2, 1]

Generator 與 yield

含 yield 的函式呼叫時不執行函式本體，而是回傳一個 generator 物件。每次 next() 執行到下一個 yield 暫停，frame（local 變數、指令位置）被保存在 heap 上，下次從暫停處繼續。這讓 generator 成為「惰性求值」與「O(1) 記憶體處理串流」的利器，也是 asyncio coroutine 的歷史基礎。

import json

def read_events(path):
    """逐行讀取上 GB 的 JSON Lines，記憶體只用一行的量"""
    with open(path) as f:
        for line in f:
            yield json.loads(line)

def only_clicks(events):
    return (e for e in events if e.get("type") == "click")   # generator expression

def batched_ids(events, size):
    batch = []
    for e in events:
        batch.append(e["user_id"])
        if len(batch) == size:
            yield batch
            batch = []
    if batch:
        yield batch

# pipeline：每一層都是惰性的，資料一筆一筆流過
# for chunk in batched_ids(only_clicks(read_events("events.jsonl")), 500):
#     bulk_write(chunk)

# list comprehension vs generator expression
import sys
print(sys.getsizeof([x for x in range(1_000_000)]))   # 約 8 MB
print(sys.getsizeof((x for x in range(1_000_000))))   # 約 200 bytes，與資料量無關
print(sum(x * x for x in range(10)))                  # 當唯一參數時可省略括號

yield from、send() 與 return 值

def inner():
    received = yield "first"          # yield 是運算式，可以接收 send 進來的值
    print("inner got", received)
    yield "second"
    return "inner-done"               # return 值放在 StopIteration.value

def outer():
    result = yield from inner()       # 委派：把 next/send/throw 直接轉給 inner，並取得其 return 值
    print("outer got", result)
    yield "third"

g = outer()
print(next(g))        # first
print(g.send("hi"))   # inner got hi → second
print(next(g))        # outer got inner-done → third

yield from 除了扁平化巢狀 generator，還正確轉發 send()、throw()、close()。Python 3.4 的 asyncio 就是用 yield from 實作 coroutine，3.5 才引入 async/await 專用語法（概念上 await ≈ yield from）。

itertools 必會清單

import itertools as it

print(list(it.islice(it.count(10, 5), 3)))              # [10, 15, 20]：無限序列 + 切片
print(list(it.chain([1, 2], (3,), {4})))                # [1, 2, 3, 4]
print(list(it.chain.from_iterable([[1], [2, 3]])))      # 攤平一層
print(list(it.accumulate([1, 2, 3, 4])))                # [1, 3, 6, 10] prefix sum
print(list(it.product("ab", repeat=2)))                 # 笛卡兒積
print(list(it.permutations([1, 2, 3], 2)))              # 排列
print(list(it.combinations([1, 2, 3], 2)))              # 組合
print(list(it.pairwise([1, 2, 4, 7])))                  # [(1,2),(2,4),(4,7)]（3.10+）
print(list(it.batched(range(7), 3)))                    # [(0,1,2),(3,4,5),(6,)]（3.12+）

# groupby 只合併「相鄰」相同 key，通常要先排序
logs = [("u1", "click"), ("u1", "view"), ("u2", "view"), ("u1", "buy")]
for user, grp in it.groupby(sorted(logs), key=lambda r: r[0]):
    print(user, [a for _, a in grp])
# u1 ['buy', 'click', 'view']
# u2 ['view']

a, b = it.tee(iter(range(3)))   # 一個 iterator 分成兩個（內部會緩衝）

Decorator 與 Context Manager 高頻

Decorator 本質：接收函式、回傳函式的 callable

@deco 放在 def f 上面等價於 f = deco(f)，在定義時（import 時）執行一次。能成立是因為函式是一級物件（first-class）加上 closure。

import functools, time, logging

def timed(func):
    @functools.wraps(func)            # 複製 __name__、__doc__、__qualname__、__wrapped__ 等
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        try:
            return func(*args, **kwargs)
        finally:
            ms = (time.perf_counter() - start) * 1000
            logging.info("%s took %.2f ms", func.__qualname__, ms)
    return wrapper

@timed
def fetch_user(uid: int) -> dict:
    """從 DB 讀使用者"""
    return {"id": uid}

print(fetch_user.__name__, fetch_user.__doc__)   # fetch_user 從 DB 讀使用者（沒 wraps 會變成 wrapper None）

為什麼一定要 functools.wraps：沒有它，被裝飾的函式名稱、docstring、型別註解都變成 wrapper 的，導致 log、除錯、FastAPI 依 signature 產生 OpenAPI、pytest fixture、inspect.signature 都出錯。

帶參數的 decorator：三層函式

import functools, random, time

def retry(times=3, backoff=0.1, exceptions=(ConnectionError, TimeoutError)):
    """@retry(times=5) → retry(times=5) 先回傳真正的 decorator，再套到函式上"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions:
                    if attempt == times:
                        raise                                  # 保留原始 traceback
                    sleep = backoff * 2 ** (attempt - 1)
                    time.sleep(sleep + random.uniform(0, sleep))  # exponential backoff + jitter
        return wrapper
    return decorator

@retry(times=5, backoff=0.05)
def call_partner_api():
    if random.random() < 0.5:
        raise ConnectionError("flaky")
    return "ok"

也可以用 class 實作 decorator（實作 __call__），適合需要保存狀態（例如計數、rate limit bucket）時。多個 decorator 疊加時，離函式最近的先套用：@a @b def f 等於 f = a(b(f))。

functools.lru_cache / cache

from functools import lru_cache, cache

@lru_cache(maxsize=1024)        # 參數必須 hashable；maxsize=None 等於 @cache（無上限）
def fib(n: int) -> int:
    return n if n < 2 else fib(n - 1) + fib(n - 2)

print(fib(200))
print(fib.cache_info())          # CacheInfo(hits=198, misses=201, maxsize=1024, currsize=201)
fib.cache_clear()

Context Manager：__enter__ / __exit__

with 保證「取得資源 → 使用 → 不論是否例外都釋放」。__exit__(exc_type, exc, tb) 回傳 truthy 值會吞掉例外（多數情況應回傳 None/False）。

import time, threading
from contextlib import contextmanager, suppress, ExitStack

class Timer:
    def __enter__(self):
        self.start = time.perf_counter()
        return self                      # as 後面拿到的東西
    def __exit__(self, exc_type, exc, tb):
        self.elapsed = time.perf_counter() - self.start
        return False                     # 不吞例外

with Timer() as t:
    sum(range(1_000_000))
print(f"{t.elapsed * 1000:.1f} ms")

@contextmanager
def acquire_with_timeout(lock: threading.Lock, timeout: float):
    ok = lock.acquire(timeout=timeout)
    if not ok:
        raise TimeoutError("lock busy")
    try:
        yield                            # with 區塊在這裡執行；區塊中的例外會在此處被 raise
    finally:
        lock.release()                   # 一定要 try/finally，否則例外時不會釋放

with suppress(FileNotFoundError):        # 取代 try/except/pass
    open("/not/exist")

# ExitStack：動態數量的資源（例如開 N 個檔案），任何一個失敗都會正確關閉已開啟的
paths = []
with ExitStack() as stack:
    files = [stack.enter_context(open(p)) for p in paths]

常見應用：DB transaction（成功 commit、例外 rollback）、鎖、暫時切換設定、計時、OpenTelemetry span。async 版本是 __aenter__/__aexit__ 搭配 async with，以及 contextlib.asynccontextmanager。

物件導向進階：descriptor、MRO、metaclass、dataclass 與 typing 深入

屬性查找順序與 descriptor

Descriptor 是實作了 __get__/__set__/__delete__ 的物件，放在 class 屬性上時會攔截 instance 的屬性存取。property、classmethod、staticmethod、函式本身（function 實作 __get__ 才變成 bound method）、__slots__ 都是 descriptor。

obj.attr 的查找順序（object.__getattribute__）：

1. 沿 MRO 在 type(obj) 找 attr；若是 data descriptor（有 __set__ 或 __delete__）→ 呼叫其 __get__。
2. 否則看 obj.__dict__ 有沒有 → 有就回傳。
3. 否則若第 1 步找到的是 non-data descriptor（只有 __get__，例如函式）→ 呼叫 __get__；是普通值就直接回傳。
4. 都沒有 → 呼叫 __getattr__（若有定義）→ 否則 AttributeError。

class Positive:
    """可重用的驗證 descriptor"""
    def __set_name__(self, owner, name):          # 3.6+：自動得知被指派到的屬性名
        self.private = "_" + name
    def __get__(self, obj, objtype=None):
        if obj is None:
            return self                           # 從 class 存取時回傳 descriptor 本身
        return getattr(obj, self.private)
    def __set__(self, obj, value):
        if value <= 0:
            raise ValueError(f"{self.private[1:]} must be > 0")
        setattr(obj, self.private, value)

class Bid:
    price = Positive()
    budget = Positive()
    def __init__(self, price, budget):
        self.price, self.budget = price, budget  # 會經過 Positive.__set__

    @property
    def max_impressions(self):                   # property 是 data descriptor 的語法糖
        return int(self.budget / self.price * 1000)

b = Bid(price=2.5, budget=100)
print(b.max_impressions)   # 40000
# Bid(price=-1, budget=1) → ValueError: price must be > 0

functools.cached_property 是 non-data descriptor：第一次計算後把結果寫入 obj.__dict__，之後依查找順序第 2 步直接命中 instance dict，不再呼叫 descriptor。

MRO、C3 linearization 與 super()

多重繼承時，Python 用 C3 linearization 計算 Method Resolution Order：保證 (1) 子類別在父類別前面；(2) 保留每個 class 宣告的父類別順序（local precedence）；(3) 單調性。無法滿足時直接 TypeError。

class Base:
    def save(self): print("Base.save")

class CacheMixin(Base):
    def save(self):
        print("CacheMixin.save: invalidate cache")
        super().save()

class AuditMixin(Base):
    def save(self):
        print("AuditMixin.save: write audit log")
        super().save()

class Campaign(CacheMixin, AuditMixin):
    def save(self):
        print("Campaign.save")
        super().save()

print([c.__name__ for c in Campaign.__mro__])
# ['Campaign', 'CacheMixin', 'AuditMixin', 'Base', 'object']
Campaign().save()
# Campaign.save → CacheMixin.save → AuditMixin.save → Base.save

關鍵：super() 不是「呼叫父類別」，而是「呼叫 MRO 中下一個 class」，而這個 MRO 是依 type(self) 決定的。所以 CacheMixin 裡的 super() 在 Campaign instance 上會呼叫到兄弟類別 AuditMixin。這就是 cooperative multiple inheritance：每個 class 都呼叫 super()，菱形繼承中 Base.save 只會被呼叫一次。無參數 super() 靠編譯器注入的 __class__ cell 與第一個參數實作。

Metaclass 簡介

class 也是物件，建立 class 的「class」就是 metaclass（預設 type）。class A: ... 大致等於 A = type("A", (bases), namespace)。Metaclass 可以在 class 被建立時介入：註冊、驗證、修改屬性。Django ORM 的 Model、SQLAlchemy declarative、舊版 pydantic 都用過 metaclass。

class RegistryMeta(type):
    registry: dict[str, type] = {}
    def __new__(mcls, name, bases, ns):
        cls = super().__new__(mcls, name, bases, ns)
        if bases:                              # 不註冊抽象基底
            mcls.registry[name.lower()] = cls
        return cls

class Handler(metaclass=RegistryMeta): ...
class ClickHandler(Handler): ...
print(RegistryMeta.registry)    # {'clickhandler': <class '__main__.ClickHandler'>}

# 大多數情況用更簡單的 __init_subclass__（3.6+）就夠了
class Plugin:
    plugins = {}
    def __init_subclass__(cls, /, key: str, **kw):
        super().__init_subclass__(**kw)
        Plugin.plugins[key] = cls
class S3Sink(Plugin, key="s3"): ...
print(Plugin.plugins)           # {'s3': <class '__main__.S3Sink'>}

dataclass

from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone

@dataclass(frozen=True, slots=True)            # frozen → 不可變且自動產生 __hash__；slots → 省記憶體（3.10+）
class BidRequest:
    request_id: str
    user_id: str
    floor_price: float
    tags: tuple[str, ...] = ()                 # 不可變預設值 OK
    ts: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

@dataclass(order=True)
class Job:
    priority: int
    name: str = field(compare=False)
    payload: list = field(default_factory=list, repr=False)   # 可變預設值必須用 default_factory

r = BidRequest("r1", "u9", 0.8)
print(r, asdict(r)["floor_price"])
print(sorted([Job(3, "c"), Job(1, "a")])[0].name)   # a

typing：Protocol、TypedDict、Generic 與 mypy

型別註解在執行期不會被強制（只存在 __annotations__），由 mypy / pyright 靜態檢查，或由 pydantic / FastAPI 在執行期讀取使用。

from typing import Protocol, TypedDict, NotRequired, TypeVar, Generic, Iterable, runtime_checkable

# 1) Protocol：structural typing（duck typing 的靜態版），不需繼承
@runtime_checkable
class Cache(Protocol):
    def get(self, key: str) -> bytes | None: ...
    def set(self, key: str, value: bytes, ttl: int) -> None: ...

class DictCache:                      # 沒有繼承 Cache，但結構相符就通過型別檢查
    def __init__(self): self._d: dict[str, bytes] = {}
    def get(self, key: str) -> bytes | None: return self._d.get(key)
    def set(self, key: str, value: bytes, ttl: int) -> None: self._d[key] = value

def warm_up(cache: Cache, items: Iterable[tuple[str, bytes]]) -> None:
    for k, v in items:
        cache.set(k, v, ttl=60)

warm_up(DictCache(), [("a", b"1")])
print(isinstance(DictCache(), Cache))   # True（runtime_checkable 只檢查方法是否存在）

# 2) TypedDict：描述 JSON / dict 的形狀
class Event(TypedDict):
    event_id: str
    user_id: str
    value: float
    campaign_id: NotRequired[str]

def handle(e: Event) -> float:
    return e["value"]          # mypy 會檢查 key 名稱與型別

# 3) Generic（3.12+ 可寫成 class Box[T]: ...，PEP 695）
T = TypeVar("T")
class Batch(Generic[T]):
    def __init__(self) -> None: self.items: list[T] = []
    def add(self, item: T) -> None: self.items.append(item)

b: Batch[int] = Batch()
b.add(1)
# b.add("x")   # mypy: Argument 1 to "add" of "Batch" has incompatible type "str"; expected "int"

uv pip install mypy
mypy --strict app/          # CI 中逐步導入：先對新模組 strict，舊模組用 per-module 設定放寬

asyncio 與並行模型比較 高頻

asyncio 是單執行緒、協作式（cooperative）多工：一個 event loop 管理大量 coroutine，coroutine 在 await 一個尚未完成的 I/O 時主動讓出控制權，loop 再用 OS 的 I/O multiplexing（Linux 上是 epoll）找出已就緒的 socket，恢復對應的 coroutine。因為切換只發生在 await 點，不需要 thread 切換成本，也幾乎不需要鎖。

核心概念

coroutine function

async def f()；呼叫 f() 只會產生 coroutine 物件，不會執行（忘記 await 會得到 "coroutine was never awaited" 警告）。

await

暫停目前 coroutine，直到 awaitable（coroutine、Task、Future）完成。只能在 async def 裡用。

Task

把 coroutine 包起來排進 event loop 並發執行（asyncio.create_task）。要保留 Task 的參考，否則可能被 GC 回收。

Future

低階的「未來結果」占位物件，Task 是它的子類別。

event loop

排程器；asyncio.run(main()) 建立 loop、執行到 main 完成、關閉 loop。

import asyncio, time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)          # 模擬網路 I/O；讓出控制權
    return f"{name} done"

async def main():
    t = time.perf_counter()
    # 1) 循序 await：總共 1 + 2 = 3 秒
    # await fetch("a", 1); await fetch("b", 2)

    # 2) gather：並發，總共約 2 秒（最慢的那個）
    results = await asyncio.gather(fetch("a", 1), fetch("b", 2), return_exceptions=True)
    print(results, f"{time.perf_counter() - t:.1f}s")

    # 3) TaskGroup（3.11+，structured concurrency）：任一失敗會取消其他並一起拋出 ExceptionGroup
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch("c", 0.5))
        t2 = tg.create_task(fetch("d", 0.3))
    print(t1.result(), t2.result())

    # 4) timeout：3.11+ asyncio.timeout；舊版 asyncio.wait_for
    try:
        async with asyncio.timeout(0.1):
            await fetch("slow", 1)
    except TimeoutError:
        print("timed out → 回傳預設值 / no-bid")

    # 5) 限制並發數：Semaphore（例如同時最多 10 個對外連線）
    sem = asyncio.Semaphore(10)
    async def limited(i):
        async with sem:
            return await fetch(str(i), 0.01)
    await asyncio.gather(*(limited(i) for i in range(100)))

asyncio.run(main())

千萬不要在 async 中呼叫 blocking code

event loop 只有一個 thread。若某個 coroutine 呼叫 time.sleep(1)、requests.get()、同步 DB driver、或做 200ms 的 CPU 運算，整個 loop 都卡住，所有其他請求的延遲一起暴增。這是 async 服務 p99 爆掉最常見的原因。

import asyncio, time, hashlib
from concurrent.futures import ProcessPoolExecutor

def blocking_io():                     # 例如舊的同步 SDK
    time.sleep(0.5); return "io"

def cpu_heavy(data: bytes) -> str:     # 例如大量 hash / 特徵計算
    for _ in range(200_000):
        data = hashlib.sha256(data).digest()
    return data.hex()[:8]

async def main():
    loop = asyncio.get_running_loop()
    # IO-bound 的同步函式 → 丟到 thread pool（3.9+ 簡寫 asyncio.to_thread）
    r1 = await asyncio.to_thread(blocking_io)
    # CPU-bound → 丟到 process pool，避開 GIL
    with ProcessPoolExecutor() as pool:
        r2 = await loop.run_in_executor(pool, cpu_heavy, b"seed")
    print(r1, r2)

if __name__ == "__main__":
    asyncio.run(main())

async context manager 與 async iterator

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def db_conn(pool):
    conn = await pool.acquire()
    try:
        yield conn
    finally:
        await pool.release(conn)

async def stream_events(n: int):      # async generator：可在 yield 之間 await
    for i in range(n):
        await asyncio.sleep(0.01)      # 例如從 Kafka 拉下一批
        yield {"id": i}

async def consume():
    async for ev in stream_events(3):
        print(ev)
    squares = [e["id"] ** 2 async for e in stream_events(3)]   # async comprehension
    print(squares)

asyncio.run(consume())

取消（cancellation）語意

task.cancel() 會在 task 下一個 await 點丟出 asyncio.CancelledError。3.8 起它繼承 BaseException，所以 except Exception 不會誤吞；若你捕捉它做清理，清理完要重新 raise。timeout 就是靠取消實作的。客戶端斷線時，ASGI server 也可能取消對應的 request handler。

threading vs multiprocessing vs asyncio vs concurrent.futures

from concurrent.futures import ThreadPoolExecutor, as_completed
import urllib.request

URLS = ["https://example.com", "https://example.org"]

def get(url):
    with urllib.request.urlopen(url, timeout=5) as r:
        return url, r.status

with ThreadPoolExecutor(max_workers=8) as ex:
    futures = {ex.submit(get, u): u for u in URLS}
    for fut in as_completed(futures):          # 誰先完成先處理
        try:
            print(fut.result())
        except Exception as e:
            print(futures[fut], "failed:", e)

例外處理最佳實踐與打包環境 進階

例外處理

import logging, json
log = logging.getLogger(__name__)

class AppError(Exception):
    """應用層例外的基底，方便上層統一處理"""
class NotFoundError(AppError): ...
class UpstreamError(AppError): ...

def load_campaign(cid: str, raw: str) -> dict:
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as e:          # 1) 捕捉具體的例外，不要裸 except
        raise AppError(f"bad payload for {cid}") from e   # 2) 用 from 保留因果鏈（__cause__）
    else:                                      # 3) 沒有例外才執行；讓 try 區塊盡量小
        if "id" not in data:
            raise NotFoundError(cid)
        return data
    finally:                                   # 4) 一定會執行：清理資源
        log.debug("load_campaign %s finished", cid)

try:
    load_campaign("c1", "{bad json")
except AppError:
    log.exception("failed")                    # 5) log.exception 會自動附上 traceback

# 6) 3.11+：ExceptionGroup 與 except*，處理並發任務的多個錯誤
try:
    raise ExceptionGroup("batch", [ValueError("a"), TypeError("b")])
except* ValueError as eg:
    print("values:", eg.exceptions)
except* TypeError as eg:
    print("types:", eg.exceptions)

• EAFP vs LBYL：Python 慣用「先做再處理例外」（Easier to Ask Forgiveness than Permission），例如直接 d[k] 再捕捉 KeyError，避免 check-then-act 的 race（檔案存在檢查後被刪除）。但例外路徑很頻繁時，dict.get 更快更清楚。
• 不要用例外控制正常流程中的高頻分支——拋出例外在 CPython 有成本（3.11 起 try 區塊本身是「zero-cost」，只有真的拋出才付費）。
• 在服務邊界（API handler、worker 主迴圈）統一捕捉，轉成正確的 HTTP 狀態碼與結構化 log；內部不要到處 catch 再吞掉。
• 重試只針對 transient 錯誤（逾時、連線重置、HTTP 503），且要有上限、backoff 與 jitter；非冪等操作要搭配 idempotency key。

打包與環境：uv、venv、pyproject.toml

現代 Python 專案以 pyproject.toml（PEP 517/518/621）為唯一設定來源，取代 setup.py + requirements.txt + setup.cfg 的組合。uv（Astral 以 Rust 撰寫）整合了 pip、venv、pip-tools、pyenv、pipx 的功能，解析與安裝速度比 pip 快一到兩個數量級。

uv venv                       # 建立 .venv
source .venv/bin/activate
uv pip install fastapi uvicorn    # pip 相容介面

# 或以專案為中心的工作流
uv init bidder && cd bidder
uv add fastapi "uvicorn[standard]" orjson
uv add --dev pytest mypy ruff
uv lock                       # 產生 uv.lock（跨平台可重現的鎖定檔）
uv sync --frozen              # CI / Docker 中依 lock 安裝，鎖定檔與 pyproject 不一致就失敗
uv run pytest
uv python install 3.13        # 管理 Python 版本本身

[project]
name = "bidder"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.115",
    "uvicorn[standard]>=0.30",
    "orjson>=3.10",
]

[dependency-groups]
dev = ["pytest>=8", "mypy>=1.11", "ruff>=0.6"]

[tool.ruff]
line-length = 100

[tool.mypy]
strict = true

Python Web 生態：WSGI、ASGI、worker 模型與框架 高頻

WSGI vs ASGI

# WSGI（PEP 3333）：同步，一個請求佔用一個 worker（thread/process）直到回應完成
def wsgi_app(environ, start_response):
    start_response("200 OK", [("Content-Type", "text/plain")])
    return [b"hello"]

# ASGI：非同步，能處理 HTTP、WebSocket、長連線；receive/send 都是 awaitable
async def asgi_app(scope, receive, send):
    assert scope["type"] == "http"
    await send({"type": "http.response.start", "status": 200,
                "headers": [(b"content-type", b"text/plain")]})
    await send({"type": "http.response.body", "body": b"hello"})

gunicorn / uvicorn 的 worker 模型

• pre-fork：master process 先 import app（若 --preload）再 fork 出 N 個 worker，共用 listening socket；master 負責監控、重啟掛掉或逾時的 worker、處理 HUP 平滑重啟。
• worker 數常見經驗值：sync worker 用 2 × cores + 1（因為會阻塞在 I/O）；async（uvicorn）worker 用約等於 cores，因為每個 worker 已能並發處理大量 I/O。最終要以壓測決定。
• --max-requests + --max-requests-jitter：處理一定數量請求後重啟 worker，緩解慢性記憶體洩漏（治標）。
• --timeout：worker 卡住超過時間被 master 殺掉重啟（sync worker 適用）。
• 多 worker 意味著：in-process 快取（lru_cache、全域 dict）每個 worker 一份、連線池大小 = 每 worker 池大小 × worker 數 × Pod 數，要算進 DB 的 max_connections。

gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000 \
  --max-requests 10000 --max-requests-jitter 1000 --timeout 30 --graceful-timeout 20
# 或直接 uvicorn（在 K8s 上常讓每個 Pod 跑 1 個 worker，由 HPA 水平擴展）
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4 --loop uvloop --http httptools

FastAPI / Flask / Django 差異

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, Field, field_validator
import asyncio

app = FastAPI()

class BidReq(BaseModel):
    request_id: str
    user_id: str
    floor_price: float = Field(gt=0, description="CPM 底價（USD）")
    slots: list[str] = Field(min_length=1)

    @field_validator("user_id")
    @classmethod
    def not_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("user_id blank")
        return v

class BidResp(BaseModel):
    request_id: str
    price: float | None

async def get_model():                  # 依賴注入：測試時可用 app.dependency_overrides 替換
    return {"base": 1.2}

@app.post("/bid", response_model=BidResp)
async def bid(req: BidReq, model: dict = Depends(get_model)) -> BidResp:
    try:
        async with asyncio.timeout(0.08):           # 給自己留 buffer，超時就 no-bid
            price = await score(req, model)
    except TimeoutError:
        return BidResp(request_id=req.request_id, price=None)
    if price < req.floor_price:
        return BidResp(request_id=req.request_id, price=None)
    return BidResp(request_id=req.request_id, price=price)

async def score(req: BidReq, model: dict) -> float:
    await asyncio.sleep(0.005)                       # 例如查 Redis 特徵
    return model["base"] * len(req.slots)
# 驗證失敗時 FastAPI 自動回 422 與詳細錯誤欄位

「為什麼 Python 會這樣」追問彙整 高頻

Go 速覽：給後端工程師的重點 進階

Appier 接受 Python / Go / Java。即使主力是 Python，懂 Go 的並行模型能讓你在系統設計題（高 QPS 服務、bidder）中更有說服力。Go 的設計目標：編譯快、部署簡單（單一靜態 binary）、原生並行、GC 停頓低。

goroutine 與 GMP 排程器（M:N）

G（goroutine）

使用者層級的輕量執行緒，初始 stack 只有約 2KB 且可動態成長，可以輕鬆開數十萬個。

M（machine）

OS thread。

P（processor）

邏輯處理器，持有一個 local run queue；數量 = GOMAXPROCS（預設為 CPU 核心數；Go 1.25 起在 Linux 上會考量 cgroup CPU limit）。M 必須綁定一個 P 才能執行 G。

• M:N：M 個 goroutine 對應到 N 個 OS thread，由 Go runtime 排程，切換成本遠低於 OS thread context switch。
• Work stealing：P 的 local queue 空了會從 global queue 或其他 P 偷一半工作。
• Syscall handoff：G 進入阻塞 syscall 時，M 連同 G 一起阻塞，P 被交給另一個 M 繼續跑其他 G。
• Network poller：網路 I/O 在底層用 epoll，G 等待 socket 時被 park，不佔用 thread——所以寫起來是同步阻塞風格，底層卻是 I/O multiplexing。
• 搶佔：Go 1.14 起有基於 signal 的非同步搶佔，緊密迴圈也不會餓死其他 goroutine。

channel、select、WaitGroup、Mutex

package main

import (
	"fmt"
	"sync"
	"time"
)

func worker(id int, jobs <-chan int, results chan<- string, wg *sync.WaitGroup) {
	defer wg.Done()
	for j := range jobs { // channel 被 close 且清空後迴圈結束
		time.Sleep(10 * time.Millisecond)
		results <- fmt.Sprintf("worker %d did job %d", id, j)
	}
}

func main() {
	jobs := make(chan int, 100)       // buffered channel
	results := make(chan string, 100)
	var wg sync.WaitGroup

	for w := 1; w <= 4; w++ { // worker pool
		wg.Add(1)
		go worker(w, jobs, results, &wg)
	}
	for j := 1; j <= 10; j++ {
		jobs <- j
	}
	close(jobs) // 只有 sender 該 close；對已 close 的 channel 送值會 panic

	go func() { wg.Wait(); close(results) }()
	for r := range results {
		fmt.Println(r)
	}

	// Mutex 保護共享狀態
	var mu sync.Mutex
	counter := map[string]int{}
	var wg2 sync.WaitGroup
	for i := 0; i < 100; i++ {
		wg2.Add(1)
		go func() {
			defer wg2.Done()
			mu.Lock()
			counter["clicks"]++
			mu.Unlock()
		}()
	}
	wg2.Wait()
	fmt.Println(counter["clicks"]) // 100
}

// select：同時等多個 channel；搭配 time.After 做逾時
select {
case res := <-resultCh:
	fmt.Println("got", res)
case <-time.After(50 * time.Millisecond):
	fmt.Println("timeout → no-bid")
}

Go 的格言：「Don't communicate by sharing memory; share memory by communicating.」但實務上兩者並用：傳遞所有權/工作流用 channel，保護一小塊狀態用 sync.Mutex / sync.RWMutex / sync/atomic。sync.Once 做一次性初始化，sync.Pool 重用暫存物件以降低 GC 壓力。

context：取消與逾時的傳遞

package main

import (
	"context"
	"errors"
	"fmt"
	"time"
)

func queryFeature(ctx context.Context, uid string) (float64, error) {
	select {
	case <-time.After(30 * time.Millisecond): // 模擬查 Redis
		return 0.42, nil
	case <-ctx.Done(): // 上游取消或逾時
		return 0, ctx.Err()
	}
}

func handleBid(parent context.Context) {
	ctx, cancel := context.WithTimeout(parent, 20*time.Millisecond)
	defer cancel() // 一定要呼叫，釋放 timer 資源
	v, err := queryFeature(ctx, "u1")
	if errors.Is(err, context.DeadlineExceeded) {
		fmt.Println("feature timeout, use default")
		return
	}
	fmt.Println("feature", v)
}

func main() { handleBid(context.Background()) }

慣例：ctx 當函式第一個參數往下傳；HTTP server 的 r.Context() 在客戶端斷線時會被取消，DB driver、HTTP client 都接受 ctx，整條呼叫鏈能一起停止，避免浪費資源（Python asyncio 的 cancellation 是類似概念）。

錯誤處理、interface、defer

package main

import (
	"errors"
	"fmt"
	"os"
)

var ErrNotFound = errors.New("not found") // sentinel error

type Store interface { // interface 是隱式實作的（structural，類似 Python Protocol）
	Get(key string) (string, error)
}

type MapStore struct{ m map[string]string }

func (s *MapStore) Get(key string) (string, error) {
	v, ok := s.m[key]
	if !ok {
		return "", fmt.Errorf("get %q: %w", key, ErrNotFound) // %w 包裝，保留錯誤鏈
	}
	return v, nil
}

func readConfig(path string) (err error) {
	f, err := os.Open(path)
	if err != nil {
		return fmt.Errorf("open config: %w", err)
	}
	defer f.Close() // defer 以 LIFO 順序在函式 return 時執行，參數在 defer 當下求值
	return nil
}

func main() {
	var s Store = &MapStore{m: map[string]string{"a": "1"}}
	if _, err := s.Get("b"); errors.Is(err, ErrNotFound) {
		fmt.Println("handled:", err) // handled: get "b": not found
	}
	_ = readConfig("/etc/hosts")
}

• 錯誤是值，不是例外：明確回傳、明確檢查，errors.Is / errors.As 檢查包裝鏈。panic/recover 只用於真正不可恢復的程式錯誤。
• nil interface 陷阱：把值為 nil 的 *MyErr 指標回傳成 error 介面，介面本身不是 nil（type 欄位有值），err != nil 為 true。

slice 底層與 append 陷阱、map 非 thread-safe

package main

import "fmt"

func main() {
	// slice = {指向底層陣列的指標, len, cap}
	a := make([]int, 3, 10) // len 3, cap 10
	b := a[:2]              // 共用同一個底層陣列
	b = append(b, 99)       // cap 足夠 → 直接寫入底層陣列，覆蓋了 a[2]！
	fmt.Println(a, b)       // [0 0 99] [0 0 99]

	c := append(a, 1) // cap 足夠 → c 與 a 共用底層陣列
	c[0] = 7
	fmt.Println(a[0]) // 7

	d := make([]int, 3, 3)
	e := append(d, 1) // cap 不足 → 配置新陣列並複製，之後 e 與 d 互不影響
	e[0] = 7
	fmt.Println(d[0]) // 0

	// 安全做法：需要獨立副本時用 full slice expression 限制 cap，或明確 copy
	safe := a[:2:2] // len 2, cap 2，之後 append 一定會重新配置
	safe = append(safe, 5)
	_ = safe
}

• map 不是 thread-safe：多個 goroutine 同時讀寫，runtime 會偵測並直接 fatal error: concurrent map writes（不是 panic，無法 recover）。用 sync.Mutex/RWMutex 包起來，或讀多寫少、key 穩定時用 sync.Map。
• map 迭代順序刻意隨機化，不能依賴。
• 用 go test -race / go run -race 開啟 race detector 抓 data race。
• goroutine 洩漏：goroutine 卡在沒人讀的 channel 上永遠不會結束；用 context 取消與 pprof 的 goroutine profile 排查。

Python vs Go：效能與並行比較

面試自測題