21103956488

Committed 18 Jan 2026 01:32AM UTC coverage: 70.588% (+1.4%) from 69.196%

Build # 21103956488

Build Type

push

github

Committed by

kimata

Commit Message

refactor: 第6回コードベースリファクタリング

主な変更点:
- webapi/: success_response/error_response ヘルパー関数を追加し統一
- cache_manager.py: 例外処理を具体化（sqlite3.Error, json.JSONDecodeError）
- cache_manager.py: _get_cache/_set_cache を private 関数化
- data_collector.py: accessor を使用した設定値アクセスに統一
- models.py: parse_row() パターンの整理
- CLAUDE.md: accessor 使用ルール、API レスポンスヘルパー等を追記
- tests/: 上記変更に対応したテストの更新

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Coverage Stats

141 of 188 new or added lines in 11 files covered. (75.0%)

5 existing lines in 2 files now uncovered.

1092 of 1547 relevant lines covered (70.59%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.8

/src/server_list/spec/cache_manager.py

#!/usr/bin/env python3
"""
Cache manager for server-list data.
Caches config, VM info, CPU benchmarks to SQLite for fast API responses.
Updates data in background and notifies via SSE.
"""

import json
import logging
import sqlite3
import threading
from datetime import datetime

import my_lib.config
import my_lib.webapp.event

import server_list.spec.db
import server_list.spec.db_config

UPDATE_INTERVAL_SEC = 300  # 5 minutes

_update_thread: threading.Thread | None = None
_should_stop = threading.Event()
_db_lock = threading.Lock()


CACHE_SCHEMA = """
CREATE TABLE IF NOT EXISTS cache (
    key TEXT PRIMARY KEY,
    value TEXT NOT NULL,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""


def init_db():
    """Initialize the cache database."""
    with server_list.spec.db.get_connection(
        server_list.spec.db_config.get_cache_db_path()
    ) as conn:
        conn.executescript(CACHE_SCHEMA)
        conn.commit()


def _get_cache(key: str) -> dict | None:
    """Internal: Get cached value by key.

    For type-safe access, use the specialized getters:
    - get_config() for config cache (returns dict | None)

    Args:
        key: Cache key to retrieve

    Returns:
        Cached value as dict, or None if not found
    """
    try:
        with _db_lock, server_list.spec.db.get_connection(
            server_list.spec.db_config.get_cache_db_path()
        ) as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT value FROM cache WHERE key = ?", (key,))
            row = cursor.fetchone()

            if row:
                return json.loads(row[0])
    except (sqlite3.Error, json.JSONDecodeError) as e:
        logging.warning("Failed to get cache for %s: %s", key, e)

    return None


def _set_cache(key: str, value: dict):
    """Internal: Set cache value."""
    try:
        with _db_lock, server_list.spec.db.get_connection(
            server_list.spec.db_config.get_cache_db_path()
        ) as conn:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT OR REPLACE INTO cache (key, value, updated_at)
                VALUES (?, ?, ?)
            """, (key, json.dumps(value, ensure_ascii=False), datetime.now().isoformat()))
            conn.commit()
    except sqlite3.Error as e:
        logging.warning("Failed to set cache for %s: %s", key, e)


def load_config_from_file() -> dict | None:
    """Load config from YAML file with schema validation.

    Uses my_lib.config.load() for consistent config loading
    with schema validation across the codebase.
    The config path is obtained from db_config for testability.
    Falls back to yaml.safe_load() if schema file is not available (e.g., in tests).
    """
    import yaml

    try:
        config_path = server_list.spec.db_config.get_config_path()
        if not config_path.exists():
            return None

        # Use centralized schema path from db module
        schema_path = server_list.spec.db.CONFIG_SCHEMA_PATH

        # Use schema validation if available, otherwise fallback to yaml.safe_load
        if schema_path.exists():
            return my_lib.config.load(config_path, schema_path)

        with open(config_path, encoding="utf-8") as f:
            return yaml.safe_load(f)
    except Exception as e:
        logging.warning("Failed to load config: %s", e)
        return None


def get_config() -> dict | None:
    """Get config from cache, or load from file if not cached.

    This is the recommended way to access config data, providing
    type-safe access with dict | None return type.

    Returns:
        Config dictionary or None if not available
    """
    cached = _get_cache("config")
    if cached:
        return cached

    # Load from file and cache
    config = load_config_from_file()
    if config:
        _set_cache("config", config)
    return config


def update_all_caches():
    """Update all caches from source data."""
    updated = False

    # Update config cache
    config = load_config_from_file()
    if config:
        old_config = _get_cache("config")
        if old_config != config:
            _set_cache("config", config)
            updated = True
            logging.info("Config cache updated")

    if updated:
        my_lib.webapp.event.notify_event(my_lib.webapp.event.EVENT_TYPE.DATA)
        logging.info("Cache updated, clients notified")


def _update_worker():
    """Background worker that updates caches periodically."""
    logging.info("Cache update worker started (interval: %d sec)", UPDATE_INTERVAL_SEC)

    # Initial update
    update_all_caches()

    while not _should_stop.wait(UPDATE_INTERVAL_SEC):
        update_all_caches()

    logging.info("Cache update worker stopped")


def start_cache_worker():
    """Start the background cache update worker."""
    global _update_thread

    init_db()

    # Initial cache population
    config = load_config_from_file()
    if config:
        _set_cache("config", config)

    if _update_thread and _update_thread.is_alive():
        return

    _should_stop.clear()
    _update_thread = threading.Thread(target=_update_worker, daemon=True)
    _update_thread.start()


def stop_cache_worker():
    """Stop the background cache update worker."""
    _should_stop.set()
    if _update_thread:
        _update_thread.join(timeout=5)

1	#!/usr/bin/env python3
2	"""
3	Cache manager for server-list data.
4	Caches config, VM info, CPU benchmarks to SQLite for fast API responses.
5	Updates data in background and notifies via SSE.
6	"""
7
8	import json	1✔
9	import logging	1✔
10	import sqlite3	1✔
11	import threading	1✔
12	from datetime import datetime	1✔
13
14	import my_lib.config	1✔
15	import my_lib.webapp.event	1✔
16
17	import server_list.spec.db	1✔
18	import server_list.spec.db_config	1✔
19
20	UPDATE_INTERVAL_SEC = 300 # 5 minutes	1✔
21
22	_update_thread: threading.Thread \| None = None	1✔
23	_should_stop = threading.Event()	1✔
24	_db_lock = threading.Lock()	1✔
25
26
27	CACHE_SCHEMA = """	1✔
28	CREATE TABLE IF NOT EXISTS cache (
29	key TEXT PRIMARY KEY,
30	value TEXT NOT NULL,
31	updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
32	)
33	"""
34
35
36	def init_db():	1✔
37	"""Initialize the cache database."""
38	with server_list.spec.db.get_connection(	1✔
39	server_list.spec.db_config.get_cache_db_path()
40	) as conn:
41	conn.executescript(CACHE_SCHEMA)	1✔
42	conn.commit()	1✔
43
44
45	def _get_cache(key: str) -> dict \| None:	1✔
46	"""Internal: Get cached value by key.
47
48	For type-safe access, use the specialized getters:
49	- get_config() for config cache (returns dict \| None)
50
51	Args:
52	key: Cache key to retrieve
53
54	Returns:
55	Cached value as dict, or None if not found
56	"""
57	try:	1✔
58	with _db_lock, server_list.spec.db.get_connection(	1✔
59	server_list.spec.db_config.get_cache_db_path()
60	) as conn:
61	cursor = conn.cursor()	1✔
62	cursor.execute("SELECT value FROM cache WHERE key = ?", (key,))	1✔
63	row = cursor.fetchone()	1✔
64
65	if row:	1✔
66	return json.loads(row[0])	1✔
67	except (sqlite3.Error, json.JSONDecodeError) as e:	1✔
68	logging.warning("Failed to get cache for %s: %s", key, e)	1✔
69
70	return None	1✔
71
72
73	def _set_cache(key: str, value: dict):	1✔
74	"""Internal: Set cache value."""
75	try:	1✔
76	with _db_lock, server_list.spec.db.get_connection(	1✔
77	server_list.spec.db_config.get_cache_db_path()
78	) as conn:
79	cursor = conn.cursor()	1✔
80	cursor.execute("""	1✔
81	INSERT OR REPLACE INTO cache (key, value, updated_at)
82	VALUES (?, ?, ?)
83	""", (key, json.dumps(value, ensure_ascii=False), datetime.now().isoformat()))
84	conn.commit()	1✔
85	except sqlite3.Error as e:	1✔
86	logging.warning("Failed to set cache for %s: %s", key, e)	1✔
87
88
89	def load_config_from_file() -> dict \| None:	1✔
90	"""Load config from YAML file with schema validation.
91
92	Uses my_lib.config.load() for consistent config loading
93	with schema validation across the codebase.
94	The config path is obtained from db_config for testability.
95	Falls back to yaml.safe_load() if schema file is not available (e.g., in tests).
96	"""
97	import yaml	1✔
98
99	try:	1✔
100	config_path = server_list.spec.db_config.get_config_path()	1✔
101	if not config_path.exists():	1✔
102	return None	1✔
103
104	# Use centralized schema path from db module
105	schema_path = server_list.spec.db.CONFIG_SCHEMA_PATH	1✔
106
107	# Use schema validation if available, otherwise fallback to yaml.safe_load
108	if schema_path.exists():	1✔
109	return my_lib.config.load(config_path, schema_path)	1✔
110
111	with open(config_path, encoding="utf-8") as f:	1✔
112	return yaml.safe_load(f)	1✔
113	except Exception as e:	1✔
114	logging.warning("Failed to load config: %s", e)	1✔
115	return None	1✔
116
117
118	def get_config() -> dict \| None:	1✔
119	"""Get config from cache, or load from file if not cached.
120
121	This is the recommended way to access config data, providing
122	type-safe access with dict \| None return type.
123
124	Returns:
125	Config dictionary or None if not available
126	"""
127	cached = _get_cache("config")	1✔
128	if cached:	1✔
129	return cached	1✔
130
131	# Load from file and cache
132	config = load_config_from_file()	1✔
133	if config:	1✔
134	_set_cache("config", config)	1✔
135	return config	1✔
136
137
138	def update_all_caches():	1✔
139	"""Update all caches from source data."""
140	updated = False	1✔
141
142	# Update config cache
143	config = load_config_from_file()	1✔
144	if config:	1✔
145	old_config = _get_cache("config")	1✔
146	if old_config != config:	1✔
147	_set_cache("config", config)	1✔
148	updated = True	1✔
149	logging.info("Config cache updated")	1✔
150
151	if updated:	1✔
152	my_lib.webapp.event.notify_event(my_lib.webapp.event.EVENT_TYPE.DATA)	1✔
153	logging.info("Cache updated, clients notified")	1✔
154
155
156	def _update_worker():	1✔
157	"""Background worker that updates caches periodically."""
158	logging.info("Cache update worker started (interval: %d sec)", UPDATE_INTERVAL_SEC)	1✔
159
160	# Initial update
161	update_all_caches()	1✔
162
163	while not _should_stop.wait(UPDATE_INTERVAL_SEC):	1✔
164	update_all_caches()	×
165
166	logging.info("Cache update worker stopped")	1✔
167
168
169	def start_cache_worker():	1✔
170	"""Start the background cache update worker."""
171	global _update_thread
172
173	init_db()	1✔
174
175	# Initial cache population
176	config = load_config_from_file()	1✔
177	if config:	1✔
NEW 178	_set_cache("config", config)	×
179
180	if _update_thread and _update_thread.is_alive():	1✔
181	return	1✔
182
183	_should_stop.clear()	1✔
184	_update_thread = threading.Thread(target=_update_worker, daemon=True)	1✔
185	_update_thread.start()	1✔
186
187
188	def stop_cache_worker():	1✔
189	"""Stop the background cache update worker."""
190	_should_stop.set()	1✔
191	if _update_thread:	1✔
192	_update_thread.join(timeout=5)	1✔

kimata / server-list / 21103956488

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous