案例:快取生命週期管理
案例:快取生命週期管理
本案例基於 .claude/lib/config_loader.py 的實際程式碼,展示如何用 Context Manager 管理快取生命週期。
先備知識
問題背景
現有設計
config_loader.py 使用全域變數實現快取:
1# 全域快取變數
2_agents_config_cache: Optional[dict] = None
3_quality_rules_cache: Optional[dict] = None
4
5def load_agents_config() -> dict:
6 """
7 載入代理人配置
8
9 使用全域快取避免重複讀取檔案。
10 """
11 global _agents_config_cache
12 if _agents_config_cache is None:
13 try:
14 _agents_config_cache = load_config("agents")
15 except FileNotFoundError:
16 _agents_config_cache = _get_default_agents_config()
17 return _agents_config_cache
18
19def load_quality_rules() -> dict:
20 """載入品質規則配置"""
21 global _quality_rules_cache
22 if _quality_rules_cache is None:
23 try:
24 _quality_rules_cache = load_config("quality_rules")
25 except FileNotFoundError:
26 _quality_rules_cache = _get_default_quality_rules()
27 return _quality_rules_cache
28
29def clear_config_cache() -> None:
30 """清除配置快取(用於測試或配置熱更新)"""
31 global _agents_config_cache, _quality_rules_cache
32 _agents_config_cache = None
33 _quality_rules_cache = None這個設計的優點
- 簡單直覺:全域變數是最簡單的快取方式
- 效能好:配置只讀取一次
- API 簡潔:呼叫者不需要管理快取
這個設計的限制
問題 1:測試難以隔離
1def test_load_agents_config():
2 # 測試 A:預設配置
3 config = load_agents_config()
4 assert "known_agents" in config
5
6def test_load_agents_config_custom():
7 # 測試 B:自訂配置檔案
8 # 問題:測試 A 的快取會影響測試 B!
9 config = load_agents_config()
10 # 可能拿到測試 A 的快取結果問題 2:快取生命週期不明確
1def process_hooks():
2 config = load_agents_config()
3 # ... 處理完成
4
5 # 問題:什麼時候應該清除快取?
6 # 如果配置檔案改了,這裡會用到舊的快取問題 3:清除快取容易忘記
1def test_something():
2 # 設定測試環境
3 os.environ["CLAUDE_PROJECT_DIR"] = "/tmp/test"
4
5 # 執行測試
6 config = load_agents_config()
7
8 # 問題:忘記清除快取!
9 # 下一個測試會用到這個測試的快取進階解決方案:Context Manager 管理快取
設計目標
- 明確的快取範圍:快取的生命週期有明確的開始和結束
- 自動清理:離開範圍時自動清除快取
- 測試友好:每個測試可以有獨立的快取
實作步驟
步驟 1:建立快取管理類別
1from contextlib import contextmanager
2from pathlib import Path
3from typing import Any, Iterator, Optional
4import os
5
6class ConfigManager:
7 """
8 配置管理器
9
10 用 Context Manager 控制快取的生命週期,
11 解決全域快取的問題。
12 """
13
14 def __init__(self, project_root: Optional[str] = None):
15 """
16 初始化配置管理器
17
18 Args:
19 project_root: 專案根目錄,預設從環境變數讀取
20 """
21 if project_root is None:
22 project_root = os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())
23 self.project_root = Path(project_root)
24 self._cache: dict[str, Any] = {}
25
26 @property
27 def config_dir(self) -> Path:
28 """配置目錄路徑"""
29 return self.project_root / ".claude" / "config"
30
31 def load_config(self, name: str) -> dict:
32 """
33 載入配置(使用快取)
34
35 Args:
36 name: 配置名稱
37
38 Returns:
39 配置內容
40 """
41 if name not in self._cache:
42 self._cache[name] = self._load_from_file(name)
43 return self._cache[name]
44
45 def _load_from_file(self, name: str) -> dict:
46 """從檔案載入配置(內部方法)"""
47 yaml_path = self.config_dir / f"{name}.yaml"
48 if yaml_path.exists():
49 import yaml
50 with open(yaml_path, "r", encoding="utf-8") as f:
51 return yaml.safe_load(f) or {}
52 raise FileNotFoundError(f"Config not found: {name}")
53
54 def clear_cache(self) -> None:
55 """清除所有快取"""
56 self._cache.clear()步驟 2:加入 Context Manager 支援
1class ConfigManager:
2 # ... 前面的程式碼 ...
3
4 @contextmanager
5 def cached_scope(self) -> Iterator["ConfigManager"]:
6 """
7 快取範圍 Context Manager
8
9 在這個範圍內,配置會被快取。
10 離開範圍時自動清除快取。
11
12 Yields:
13 ConfigManager: 自己,方便鏈式呼叫
14
15 Example:
16 manager = ConfigManager()
17 with manager.cached_scope():
18 config = manager.load_config("agents")
19 # ... 使用配置 ...
20 # 離開時快取自動清除
21 """
22 try:
23 yield self
24 finally:
25 self.clear_cache()
26
27 @contextmanager
28 def isolated_scope(
29 self,
30 project_root: Optional[str] = None
31 ) -> Iterator["ConfigManager"]:
32 """
33 隔離範圍 Context Manager
34
35 建立一個完全隔離的配置環境,
36 適合用於測試。
37
38 Args:
39 project_root: 臨時的專案根目錄
40
41 Yields:
42 ConfigManager: 新的管理器實例
43
44 Example:
45 with manager.isolated_scope("/tmp/test") as isolated:
46 config = isolated.load_config("agents")
47 # 完全隔離,不影響原本的 manager
48 """
49 isolated_manager = ConfigManager(
50 project_root=project_root or str(self.project_root)
51 )
52 try:
53 yield isolated_manager
54 finally:
55 isolated_manager.clear_cache()步驟 3:加入便利方法
1class ConfigManager:
2 # ... 前面的程式碼 ...
3
4 def load_agents_config(self) -> dict:
5 """載入代理人配置(帶預設值)"""
6 try:
7 return self.load_config("agents")
8 except FileNotFoundError:
9 return self._get_default_agents_config()
10
11 def load_quality_rules(self) -> dict:
12 """載入品質規則配置(帶預設值)"""
13 try:
14 return self.load_config("quality_rules")
15 except FileNotFoundError:
16 return self._get_default_quality_rules()
17
18 def _get_default_agents_config(self) -> dict:
19 """預設代理人配置"""
20 return {
21 "known_agents": ["basil", "thyme", "mint"],
22 "agent_dispatch_rules": {},
23 }
24
25 def _get_default_quality_rules(self) -> dict:
26 """預設品質規則配置"""
27 return {
28 "trigger_conditions": {"allowed_tools": ["Write", "Edit"]},
29 "cache": {"ttl_minutes": 5},
30 }完整程式碼
1#!/usr/bin/env python3
2"""
3快取生命週期管理 - 完整範例
4
5展示如何用 Context Manager 管理配置快取的生命週期。
6"""
7
8from contextlib import contextmanager
9from pathlib import Path
10from typing import Any, Iterator, Optional
11import os
12
13try:
14 import yaml
15 HAS_YAML = True
16except ImportError:
17 HAS_YAML = False
18 import json
19
20class ConfigManager:
21 """
22 配置管理器
23
24 用 Context Manager 控制快取的生命週期。
25 """
26
27 def __init__(self, project_root: Optional[str] = None):
28 if project_root is None:
29 project_root = os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())
30 self.project_root = Path(project_root)
31 self._cache: dict[str, Any] = {}
32
33 @property
34 def config_dir(self) -> Path:
35 return self.project_root / ".claude" / "config"
36
37 # ===== 載入方法 =====
38
39 def load_config(self, name: str) -> dict:
40 """載入配置(使用快取)"""
41 if name not in self._cache:
42 self._cache[name] = self._load_from_file(name)
43 return self._cache[name]
44
45 def _load_from_file(self, name: str) -> dict:
46 """從檔案載入配置"""
47 yaml_path = self.config_dir / f"{name}.yaml"
48 json_path = self.config_dir / f"{name}.json"
49
50 if yaml_path.exists() and HAS_YAML:
51 with open(yaml_path, "r", encoding="utf-8") as f:
52 return yaml.safe_load(f) or {}
53
54 if json_path.exists():
55 with open(json_path, "r", encoding="utf-8") as f:
56 return json.load(f)
57
58 raise FileNotFoundError(f"Config not found: {name}")
59
60 def clear_cache(self) -> None:
61 """清除所有快取"""
62 self._cache.clear()
63
64 # ===== Context Manager =====
65
66 @contextmanager
67 def cached_scope(self) -> Iterator["ConfigManager"]:
68 """快取範圍:離開時自動清除快取"""
69 try:
70 yield self
71 finally:
72 self.clear_cache()
73
74 @contextmanager
75 def isolated_scope(
76 self,
77 project_root: Optional[str] = None
78 ) -> Iterator["ConfigManager"]:
79 """隔離範圍:建立獨立的配置環境"""
80 isolated = ConfigManager(project_root or str(self.project_root))
81 try:
82 yield isolated
83 finally:
84 isolated.clear_cache()
85
86 # ===== 便利方法 =====
87
88 def load_agents_config(self) -> dict:
89 """載入代理人配置"""
90 try:
91 return self.load_config("agents")
92 except FileNotFoundError:
93 return {"known_agents": [], "agent_dispatch_rules": {}}
94
95 def load_quality_rules(self) -> dict:
96 """載入品質規則配置"""
97 try:
98 return self.load_config("quality_rules")
99 except FileNotFoundError:
100 return {"trigger_conditions": {}, "cache": {"ttl_minutes": 5}}
101
102# ===== 全域便利函式(相容舊 API)=====
103
104_default_manager: Optional[ConfigManager] = None
105
106def get_config_manager() -> ConfigManager:
107 """獲取預設的配置管理器"""
108 global _default_manager
109 if _default_manager is None:
110 _default_manager = ConfigManager()
111 return _default_manager
112
113def load_agents_config() -> dict:
114 """相容舊 API:載入代理人配置"""
115 return get_config_manager().load_agents_config()
116
117def load_quality_rules() -> dict:
118 """相容舊 API:載入品質規則配置"""
119 return get_config_manager().load_quality_rules()
120
121@contextmanager
122def config_scope(project_root: Optional[str] = None) -> Iterator[ConfigManager]:
123 """
124 配置範圍 Context Manager
125
126 建立一個有明確生命週期的配置環境。
127
128 Args:
129 project_root: 專案根目錄
130
131 Yields:
132 ConfigManager: 配置管理器
133
134 Example:
135 with config_scope("/path/to/project") as config:
136 agents = config.load_agents_config()
137 rules = config.load_quality_rules()
138 # 離開時快取自動清除
139 """
140 manager = ConfigManager(project_root)
141 with manager.cached_scope():
142 yield manager
143
144# ===== 測試範例 =====
145
146if __name__ == "__main__":
147 import tempfile
148
149 # 建立測試配置
150 with tempfile.TemporaryDirectory() as tmpdir:
151 config_dir = Path(tmpdir) / ".claude" / "config"
152 config_dir.mkdir(parents=True)
153
154 # 寫入測試配置
155 (config_dir / "agents.json").write_text(
156 '{"known_agents": ["test-agent-1", "test-agent-2"]}'
157 )
158
159 # 使用 Context Manager
160 with config_scope(tmpdir) as config:
161 agents = config.load_agents_config()
162 print(f"代理人配置: {agents}")
163
164 # 第二次呼叫會使用快取
165 agents2 = config.load_agents_config()
166 print(f"快取命中: {agents is agents2}")
167
168 # 離開範圍後快取已清除
169 print("離開範圍,快取已清除")
170
171 # 測試隔離範圍
172 print("\n=== 測試隔離範圍 ===")
173 manager = ConfigManager()
174
175 with manager.isolated_scope() as isolated:
176 # 隔離環境
177 try:
178 config = isolated.load_config("nonexistent")
179 except FileNotFoundError as e:
180 print(f"預期的錯誤: {e}")
181
182 print("隔離範圍結束")使用範例
基本使用
1# 建立配置管理器
2manager = ConfigManager()
3
4# 使用快取範圍
5with manager.cached_scope():
6 agents = manager.load_agents_config()
7 rules = manager.load_quality_rules()
8
9 # 第二次呼叫會使用快取
10 agents2 = manager.load_agents_config()
11 assert agents is agents2 # 同一個物件
12
13# 離開範圍後快取已清除測試中使用
1import pytest
2from pathlib import Path
3import tempfile
4
5@pytest.fixture
6def config_manager():
7 """測試用的配置管理器"""
8 with tempfile.TemporaryDirectory() as tmpdir:
9 # 準備測試配置
10 config_dir = Path(tmpdir) / ".claude" / "config"
11 config_dir.mkdir(parents=True)
12 (config_dir / "agents.json").write_text(
13 '{"known_agents": ["test-agent"]}'
14 )
15
16 # 使用隔離範圍
17 manager = ConfigManager()
18 with manager.isolated_scope(tmpdir) as isolated:
19 yield isolated
20
21def test_load_agents_config(config_manager):
22 """測試載入代理人配置"""
23 config = config_manager.load_agents_config()
24 assert config["known_agents"] == ["test-agent"]
25
26def test_cache_works(config_manager):
27 """測試快取功能"""
28 config1 = config_manager.load_agents_config()
29 config2 = config_manager.load_agents_config()
30 assert config1 is config2 # 同一個物件相容舊 API
1# 如果有既有程式碼使用舊 API
2from config_loader import load_agents_config, config_scope
3
4# 舊的呼叫方式仍然可用
5config = load_agents_config()
6
7# 新的呼叫方式有明確的生命週期
8with config_scope() as manager:
9 config = manager.load_agents_config()設計權衡
| 面向 | 全域快取 | Context Manager |
|---|---|---|
| 簡單性 | 最簡單 | 需要理解 Context Manager |
| 生命週期 | 不明確(程式結束前一直存在) | 明確(範圍結束時清除) |
| 測試隔離 | 困難(需要手動清除) | 容易(自動隔離) |
| 並行安全 | 不安全(共用全域變數) | 可以安全(每個範圍獨立) |
| 記憶體管理 | 可能洩漏 | 自動清理 |
| API 相容性 | N/A | 可以相容舊 API |
什麼時候該用 Context Manager 管理快取?
適合使用:
- 需要明確的快取生命週期
- 測試需要隔離的快取
- 可能並行存取快取
- 快取資料量大,需要及時釋放
不建議使用:
- 快取需要跨多個函式呼叫共享
- 快取很小,不需要特別管理
- 程式很簡單,不需要測試隔離
進階:與 ExitStack 結合
當需要管理多個快取時:
1from contextlib import ExitStack
2
3def process_with_multiple_caches():
4 """使用 ExitStack 管理多個快取"""
5 with ExitStack() as stack:
6 # 進入多個快取範圍
7 config = stack.enter_context(config_scope())
8 db_cache = stack.enter_context(database_cache_scope())
9 api_cache = stack.enter_context(api_cache_scope())
10
11 # 使用各種快取
12 agents = config.load_agents_config()
13 users = db_cache.get_users()
14 data = api_cache.fetch_data()
15
16 return process(agents, users, data)
17
18 # ExitStack 會自動清理所有快取練習
基礎練習
- 為
ConfigManager加入reload_config方法,強制重新載入配置 - 實作一個
@cached_config裝飾器,讓函式自動使用快取範圍
進階練習
- 加入 TTL(Time To Live)支援,讓快取自動過期
- 實作一個執行緒安全的
ConfigManager,支援多執行緒存取
挑戰題
- 參考
config_loader.py的 Fallback Pattern(YAML → JSON → 預設值),用 Context Manager 實現「配置來源優先級」的管理