21093103049

Committed 17 Jan 2026 10:45AM UTC coverage: 68.921% (+10.4%) from 58.557%

Build # 21093103049

Build Type

push

github

Committed by

kimata

Commit Message

feat: データ構造定義とDB設定管理モジュールを追加

- models.py: VMInfo, HostInfo, PowerInfo等のdataclassを定義
- db_config.py: データベースパスの一元管理（getter/setter パターン）

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

Coverage Stats

117 of 127 new or added lines in 2 files covered. (92.13%)

273 existing lines in 8 files now uncovered.

1060 of 1538 relevant lines covered (68.92%)

0.69 hits per line

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

98.81

/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 threading
from datetime import datetime

import yaml

import my_lib.webapp.event

from server_list.spec.db import CONFIG_PATH, get_connection
from server_list.spec.db_config import get_cache_db_path

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 get_connection(get_cache_db_path()) as conn:
        conn.executescript(CACHE_SCHEMA)
        conn.commit()


def get_cache(key: str) -> dict | list | None:
    """Get cached value by key (internal use).

    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 list, or None if not found
    """
    try:
        with _db_lock, get_connection(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 Exception as e:
        logging.warning("Failed to get cache for %s: %s", key, e)

    return None


def set_cache(key: str, value: dict | list):
    """Set cache value."""
    try:
        with _db_lock, get_connection(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 Exception 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."""
    try:
        if CONFIG_PATH.exists():
            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 and isinstance(cached, dict):
        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 threading	1✔
11	from datetime import datetime	1✔
12
13	import yaml	1✔
14
15	import my_lib.webapp.event	1✔
16
17	from server_list.spec.db import CONFIG_PATH, get_connection	1✔
18	from server_list.spec.db_config import get_cache_db_path	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 get_connection(get_cache_db_path()) as conn:	1✔
39	conn.executescript(CACHE_SCHEMA)	1✔
40	conn.commit()	1✔
41
42
43	def get_cache(key: str) -> dict \| list \| None:	1✔
44	"""Get cached value by key (internal use).
45
46	For type-safe access, use the specialized getters:
47	- get_config() for config cache (returns dict \| None)
48
49	Args:
50	key: Cache key to retrieve
51
52	Returns:
53	Cached value as dict or list, or None if not found
54	"""
55	try:	1✔
56	with _db_lock, get_connection(get_cache_db_path()) as conn:	1✔
57	cursor = conn.cursor()	1✔
58	cursor.execute("SELECT value FROM cache WHERE key = ?", (key,))	1✔
59	row = cursor.fetchone()	1✔
60
61	if row:	1✔
62	return json.loads(row[0])	1✔
63	except Exception as e:	1✔
64	logging.warning("Failed to get cache for %s: %s", key, e)	1✔
65
66	return None	1✔
67
68
69	def set_cache(key: str, value: dict \| list):	1✔
70	"""Set cache value."""
71	try:	1✔
72	with _db_lock, get_connection(get_cache_db_path()) as conn:	1✔
73	cursor = conn.cursor()	1✔
74	cursor.execute("""	1✔
75	INSERT OR REPLACE INTO cache (key, value, updated_at)
76	VALUES (?, ?, ?)
77	""", (key, json.dumps(value, ensure_ascii=False), datetime.now().isoformat()))
78	conn.commit()	1✔
79	except Exception as e:	1✔
80	logging.warning("Failed to set cache for %s: %s", key, e)	1✔
81
82
83	def load_config_from_file() -> dict \| None:	1✔
84	"""Load config from YAML file."""
85	try:	1✔
86	if CONFIG_PATH.exists():	1✔
87	with open(CONFIG_PATH, encoding="utf-8") as f:	1✔
88	return yaml.safe_load(f)	1✔
89	except Exception as e:	1✔
90	logging.warning("Failed to load config: %s", e)	1✔
91	return None	1✔
92
93
94	def get_config() -> dict \| None:	1✔
95	"""Get config from cache, or load from file if not cached.
96
97	This is the recommended way to access config data, providing
98	type-safe access with dict \| None return type.
99
100	Returns:
101	Config dictionary or None if not available
102	"""
103	cached = get_cache("config")	1✔
104	if cached and isinstance(cached, dict):	1✔
105	return cached	1✔
106
107	# Load from file and cache
108	config = load_config_from_file()	1✔
109	if config:	1✔
110	set_cache("config", config)	1✔
111	return config	1✔
112
113
114	def update_all_caches():	1✔
115	"""Update all caches from source data."""
116	updated = False	1✔
117
118	# Update config cache
119	config = load_config_from_file()	1✔
120	if config:	1✔
121	old_config = get_cache("config")	1✔
122	if old_config != config:	1✔
123	set_cache("config", config)	1✔
124	updated = True	1✔
125	logging.info("Config cache updated")	1✔
126
127	if updated:	1✔
128	my_lib.webapp.event.notify_event(my_lib.webapp.event.EVENT_TYPE.DATA)	1✔
129	logging.info("Cache updated, clients notified")	1✔
130
131
132	def _update_worker():	1✔
133	"""Background worker that updates caches periodically."""
134	logging.info("Cache update worker started (interval: %d sec)", UPDATE_INTERVAL_SEC)	1✔
135
136	# Initial update
137	update_all_caches()	1✔
138
139	while not _should_stop.wait(UPDATE_INTERVAL_SEC):	1✔
UNCOV 140	update_all_caches()	×
141
142	logging.info("Cache update worker stopped")	1✔
143
144
145	def start_cache_worker():	1✔
146	"""Start the background cache update worker."""
147	global _update_thread
148
149	init_db()	1✔
150
151	# Initial cache population
152	config = load_config_from_file()	1✔
153	if config:	1✔
154	set_cache("config", config)	1✔
155
156	if _update_thread and _update_thread.is_alive():	1✔
157	return	1✔
158
159	_should_stop.clear()	1✔
160	_update_thread = threading.Thread(target=_update_worker, daemon=True)	1✔
161	_update_thread.start()	1✔
162
163
164	def stop_cache_worker():	1✔
165	"""Stop the background cache update worker."""
166	_should_stop.set()	1✔
167	if _update_thread:	1✔
168	_update_thread.join(timeout=5)	1✔

kimata / server-list / 21093103049

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