Nicole's Python Field Manual

黃底 = skeleton,直接抄,唔使改
灰色斜體 = 因應題目改嘅部分(Bank System 做例子)

Imports(最頂 copy)

import asyncio
import copy
import bisect
from collections import defaultdict

L1 — CRUD

__init__(完整 example)

class BankSystem:
    def __init__(self):
        self.accounts = {}                          # ← 主要 data store
        self.payment_counter = 0                    # ← L3 用
        self.payment = {}                            # ← L3 用
        self.merged_accounts = {}                    # ← L4 用
        self.account_locks = defaultdict(asyncio.Lock) # ← L5-6 用

create(完整 example)

def create_account(self, timestamp, account_id):
    self._process_cashbacks(timestamp)             # ← 固定第一行
    if account_id in self.accounts:
        return False
    self.accounts[account_id] = {"balance": 0, "outgoing": 0, "history": [(timestamp, 0)]}
    return True

get / deposit(完整 example)

def deposit(self, timestamp, account_id, amount):
    self._process_cashbacks(timestamp)
    if account_id not in self.accounts:
        return None
    acc = self.accounts[account_id]              # ← 攞短名
    acc["balance"] += amount                      # ← 做嘢
    acc["history"].append((timestamp, acc["balance"])) # ← 記 history
    return acc["balance"]

transfer / 雙 entity 操作(完整 example)

def transfer(self, timestamp, source_id, target_id, amount):
    self._process_cashbacks(timestamp)
    if source_id not in self.accounts or target_id not in self.accounts:
        return None
    if source_id == target_id:
        return None
    s_acc = self.accounts[source_id]
    t_acc = self.accounts[target_id]
    if s_acc["balance"] < amount:
        return None
    s_acc["balance"] -= amount
    s_acc["outgoing"] += amount
    s_acc["history"].append((timestamp, s_acc["balance"]))
    t_acc["balance"] += amount
    t_acc["history"].append((timestamp, t_acc["balance"]))
    return s_acc["balance"]
Dict vs Dot: d["balance"] 唔係 d.balance
Variable vs String: d[account_id] 用 variable,d["account_id"] 搵 literal string
Dict 係 pointer: acc = self.accounts[id] 改 acc 會改原本

L2 — Sort / Filter / Format

排序 + format(完整 example)

def top_spenders(self, timestamp, n):
    self._process_cashbacks(timestamp)
    sorted_items = dict(sorted(
        self.accounts.items(),
        key=lambda x: (-x[1]["outgoing"], x[0])     # ← desc by value, asc by name
    ))
    result = []
    for account_id, acc in sorted_items.items():
        result.append(f"{account_id}({acc['outgoing']})")
    return result[:n]                              # ← [:n] 自動處理 n 太大
sorted() 人話:「將呢堆嘢,按呢個準則排」。負號 = 反轉。
x[0] = key("alice"),x[1] = value(成個 dict)
f-string: f"{name}({value})""alice(500)"

L3 — Timestamp / TTL / Lazy Processing

Timestamp = 「而家幾點」,數字,大 = 遲
TTL = 幾耐之後過期,expiry = timestamp + ttl
Lazy processing = 唔會自動到期,要等有人 call method 先 check

Lazy Processing Helper(完整 example)

所有題都用。唔會自動到期,每次有人 call method 帶 timestamp 入嚟先 check。

def _process_cashbacks(self, timestamp):
    for pid, p in self.payment.items():
        if not p["received"] and timestamp >= p["cashback_time"]:
            self.accounts[p["account_id"]]["balance"] += p["cashback_amount"]
            p["received"] = True

pay / 建立 scheduled event(完整 example)

def pay(self, timestamp, account_id, amount):
    self._process_cashbacks(timestamp)
    if account_id not in self.accounts: return None
    acc = self.accounts[account_id]
    if acc["balance"] < amount: return None
    acc["balance"] -= amount
    acc["outgoing"] += amount
    acc["history"].append((timestamp, acc["balance"]))
    cashback = amount * 2 // 100                   # ← 計算 scheduled amount
    cashback_time = timestamp + 86400000             # ← 計算 deadline
    self.payment_counter += 1
    pid = f"payment{self.payment_counter}"
    self.payment[pid] = {"account_id": account_id, "cashback_amount": cashback, "cashback_time": cashback_time, "received": False}
    return pid

get status(完整 example)

def get_payment_status(self, timestamp, account_id, payment_id):
    self._process_cashbacks(timestamp)
    if account_id not in self.accounts: return None
    if payment_id not in self.payment: return None
    if self.payment[payment_id]["account_id"] != account_id: return None
    if self.payment[payment_id]["received"]:
        return "CASHBACK_RECEIVED"
    return "IN_PROGRESS"

L4 — Backup / Merge / Historical Query

歷史查詢(完整 example)

def get_balance(self, timestamp, account_id, time_at):
    self._process_cashbacks(timestamp)
    if account_id not in self.accounts: return None
    acc = self.accounts[account_id]
    for ts, val in reversed(acc["history"]):      # ← 由尾搵返去
        if ts <= time_at:
            return val

merge / 合併兩個 entity(完整 example)

def merge_accounts(self, timestamp, account_id_1, account_id_2):
    self._process_cashbacks(timestamp)
    if account_id_1 not in self.accounts or account_id_2 not in self.accounts: return False
    if account_id_1 == account_id_2: return False
    acc_1 = self.accounts[account_id_1]
    acc_2 = self.accounts[account_id_2]
    acc_1["balance"] += acc_2["balance"]              # ← 合併 data
    acc_1["outgoing"] += acc_2["outgoing"]
    acc_1["history"].append((timestamp, acc_1["balance"]))
    self.merged_accounts[account_id_2] = account_id_1
    del self.accounts[account_id_2]
    return True

backup / deepcopy(完整 example)

import copy
snapshot = copy.deepcopy(self.accounts)          # ← 完整複製
# 改 snapshot 唔影響原本

L5 — Concurrent Batch Processing

完整 example

三步:1. 定義 execute_op 2. gather 同時跑 3. return list(results)

async def process_batch(self, timestamp, operations):
    self._process_cashbacks(timestamp)

    async def execute_op(op):
        aid = op.get("account_id") or op.get("source_id")
        lock = self.account_locks[aid]
        async with lock:
            # -------- 以下因應題目改 --------
            if op["type"] == "deposit":
                return self.deposit(timestamp, op["account_id"], op["amount"])
            elif op["type"] == "transfer":
                return self.transfer(timestamp, op["source_id"], op["target_id"], op["amount"])
            elif op["type"] == "pay":
                return self.pay(timestamp, op["account_id"], op["amount"])
            # -------- 以上因應題目改 --------

    results = await asyncio.gather(*[execute_op(op) for op in operations])
    return list(results)

L6 — Rate Limited External Call

完整 example

同 L5 一樣,加 Semaphore。Lock 包住改 data,Sem 包住外部 call。唔好 return 太早。

async def process_external_transfers(self, timestamp, transfers, max_concurrent):
    self._process_cashbacks(timestamp)
    sem = asyncio.Semaphore(max_concurrent)

    async def do_transfer(t):
        aid = t["account_id"]
        lock = self.account_locks[aid]
        async with lock:                               # ← 鎖住 account
            # -------- 以下因應題目改(check + 扣錢) --------
            if aid not in self.accounts:
                return False
            if self.accounts[aid]["balance"] < t["amount"]:
                return False
            self.accounts[aid]["balance"] -= t["amount"]
            self.accounts[aid]["outgoing"] += t["amount"]
            self.accounts[aid]["history"].append((timestamp, self.accounts[aid]["balance"]))
            # -------- 以上因應題目改 --------
        async with sem:                                # ← 限制同時數量
            await asyncio.sleep(0.01)                  # ← 模擬外部 call
        return True

    results = await asyncio.gather(*[do_transfer(t) for t in transfers])
    return list(results)
L5 vs L6 分別:
L5:gather + Lock(全部同時跑,鎖住 account)
L6:gather + Lock + Semaphore(限制同時幾多個 + 模擬外部 call)

Lock 同 Sem 分開用。Lock 包住改 data,Sem 包住外部 call。

Gotchas(做過嘅錯誤)

_init_ 一個底線 → __init__ 兩個底線
def method(account_id): 漏 self → def method(self, account_id):
d.balance dot → d["balance"] bracket
accounts.append(acc) accounts[key] = acc dict 用 key
balance <= amount return None → balance < amount 等於夠錢要畀轉
for x in dict: x.key x 已經係 key string
payment = d[pid]: 多咗冒號 → 唔係 function 唔使冒號
return IN_PROGRESS return "IN_PROGRESS" 要引號
acc.amount = amount 覆蓋 → acc["balance"] += amount 累加
邊 iterate 邊 delete dict → 先收集 keys 再 delete
copy_ref = original 同一個 object → copy.deepcopy(original)
async with lock: return ... 之後嘅 code 行唔到 → 唔好 return 太早,lock 同 sem 分開

Error Message 點讀

Error意思點修
TypeError: takes N args but M givenparam 數量錯check 有冇 self / 有冇漏 timestamp
AttributeError: 'dict' has no attribute 'X'用咗 dot access dict改做 d["X"]
KeyError: 'X'dict 冇呢個 key先 check if key in d
AssertionError: None is not truefunction 冇 return加 return True/False
NameError: name 'X' not definedvariable 未定義check 串字 / scope

口訣:1. 睇 Error 類型 2. 睇 > 嗰行(test 點 call 你) 3. 對比你嘅 code

Quick Reference

要做咩Code
建空 dictd = {}
加/改d[key] = value
del d[key]
check 存在if key in d:
攞 value(safe)d.get(key)
loop dictfor k, v in d.items():
建空 listlst = []
加去尾lst.append(x)
頭 n 個lst[:n]
format stringf"{name}({value})"
join list", ".join(lst)
sort desc+ascsorted(items, key=lambda x: (-x[1], x[0]))
deep copycopy.deepcopy(data)
reversed loopfor ts, val in reversed(lst):
auto IDself.counter += 1; id = f"item{self.counter}"
round downamount * 2 // 100
同時跑await asyncio.gather(*[fn(x) for x in items])
鎖住async with lock:
限制數量sem = asyncio.Semaphore(n); async with sem:
Built by Nicole · github.com/ithiria894