Coverage for src / beautyspot / db.py: 74%
415 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-03-10 23:54 +0900
1# src/beautyspot/db.py
3import sqlite3
4import os
5import logging
6import queue
7import threading
8import time
9from collections.abc import Iterator
10from contextlib import contextmanager
11import dataclasses
12from dataclasses import dataclass
13from datetime import datetime, timezone
14from pathlib import Path
15from abc import ABC, abstractmethod
16from typing import Optional, TYPE_CHECKING, Any, Callable, Protocol, runtime_checkable
17import weakref
18from beautyspot.types import TaskRecord
21@runtime_checkable
22class TaskDBCore(Protocol):
23 """
24 Core interface for task metadata storage required during execution.
25 """
27 def init_schema(self) -> None: ...
29 def get(
30 self, cache_key: str, *, include_expired: bool = False
31 ) -> Optional[TaskRecord]: ...
33 def save(
34 self,
35 cache_key: str,
36 func_name: str,
37 func_identifier: Optional[str],
38 input_id: str,
39 version: Optional[str],
40 result_type: str,
41 content_type: Optional[str],
42 result_value: Optional[str] = None,
43 result_data: Optional[bytes] = None,
44 expires_at: Optional[datetime] = None,
45 ) -> None: ...
47 def delete(self, cache_key: str) -> bool: ...
50@runtime_checkable
51class Flushable(Protocol):
52 """Protocol for backends that support flushing pending writes."""
54 def flush(self, timeout: Optional[float] = None) -> bool: ...
57@runtime_checkable
58class Shutdownable(Protocol):
59 """Protocol for backends that require graceful shutdown."""
61 def shutdown(self, wait: bool = True) -> None: ...
64@runtime_checkable
65class Maintenable(Protocol):
66 """
67 Extended interface for maintenance tasks (GC, CLI, Dashboard).
68 """
70 def delete_expired(self) -> int: ...
72 def prune(self, older_than: datetime, func_name: Optional[str] = None) -> int: ...
74 def get_outdated_tasks(
75 self, older_than: datetime, func_name: Optional[str] = None
76 ) -> list[tuple[str, str, str]]: ...
78 def get_blob_refs(self) -> Optional[set[str]]: ...
80 def delete_all(self, func_name: Optional[str] = None) -> int: ...
82 def get_keys_start_with(self, prefix: str) -> list[str]: ...
84 def get_history(self, limit: int = 1000) -> "pd.DataFrame": ...
87@runtime_checkable
88class TaskDBMaintenable(TaskDBCore, Maintenable, Protocol):
89 ...
93class _ReadConnWrapper:
94 def __init__(self, conn: sqlite3.Connection):
95 self.conn = conn
96 self.lock = threading.RLock()
97 self._closed = False
98 self._shutdown_requested = False
100 def close(self, wait: bool = True):
101 """
102 Args:
103 wait: True の場合はロック解放を待機。
104 False (シャットダウン時) の場合は即座に試行し、他が使用中ならスキップする。
105 """
106 # wait=False の場合は blocking=False になり、取得できなければ直ちに False を返す
107 if not self.lock.acquire(blocking=wait):
108 # 誰かがクエリ実行中なので、強制クローズによるクラッシュを防ぐために諦める。
109 # 代わりにシャットダウン要求フラグを立てて、クエリ完了後に自身でクローズさせる。
110 if not wait: 110 ↛ 112line 110 didn't jump to line 112 because the condition on line 110 was always true
111 self._shutdown_requested = True
112 return
114 try:
115 if not self._closed:
116 try:
117 self.conn.close()
118 except Exception:
119 pass
120 self._closed = True
121 finally:
122 self.lock.release()
124 def __del__(self):
125 self.close()
128if TYPE_CHECKING:
129 import pandas as pd
131logger = logging.getLogger(__name__)
132logger.addHandler(logging.NullHandler())
135def _ensure_utc_isoformat(dt: Optional[datetime]) -> Optional[str]:
136 """datetime を UTC 保証の ISO 8601 文字列に変換する。None はそのまま返す。"""
137 if dt is None:
138 return None
139 if dt.tzinfo is None: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 dt = dt.replace(tzinfo=timezone.utc)
141 else:
142 dt = dt.astimezone(timezone.utc)
143 return dt.isoformat(" ")
146@dataclass
147class _WriteTask:
148 fn: Callable[[sqlite3.Connection], Any]
149 event: threading.Event
150 result: Any = None
151 error: Exception | BaseException | None = None
152 state: str = "PENDING" # "PENDING", "RUNNING", "DONE", "CANCELLED"
153 _state_lock: threading.Lock = dataclasses.field(default_factory=threading.Lock)
155 def try_cancel(self) -> bool:
156 """PENDING 状態のタスクをキャンセルする。成功時 True。"""
157 with self._state_lock:
158 if self.state == "PENDING":
159 self.state = "CANCELLED"
160 return True
161 return False
163 def try_start(self) -> bool:
164 """PENDING → RUNNING に遷移する。CANCELLED なら False を返す。"""
165 with self._state_lock:
166 if self.state == "CANCELLED":
167 return False
168 self.state = "RUNNING"
169 return True
171 def mark_done(self) -> None:
172 """RUNNING → DONE に遷移する。"""
173 with self._state_lock:
174 if self.state != "CANCELLED": 174 ↛ exitline 174 didn't jump to the function exit
175 self.state = "DONE"
178_STOP = object()
181class TaskDBBase(ABC):
182 """
183 Abstract base class providing default no-op implementations for maintenance methods.
184 Actual backends should implement TaskDBCore and optionally TaskDBMaintenance.
185 """
187 @abstractmethod
188 def init_schema(self):
189 pass
191 @abstractmethod
192 def get(
193 self, cache_key: str, *, include_expired: bool = False
194 ) -> Optional[TaskRecord]:
195 pass
197 @abstractmethod
198 def save(
199 self,
200 cache_key: str,
201 func_name: str,
202 func_identifier: Optional[str],
203 input_id: str,
204 version: Optional[str],
205 result_type: str,
206 content_type: Optional[str],
207 result_value: Optional[str] = None,
208 result_data: Optional[bytes] = None,
209 expires_at: Optional[datetime] = None,
210 ):
211 pass
213 @abstractmethod
214 def delete(self, cache_key: str) -> bool:
215 pass
217 # --- Maintenance Methods (Default implementations) ---
218 def delete_expired(self) -> int:
219 """Delete tasks that have passed their expiration time."""
220 return 0
222 def prune(self, older_than: datetime, func_name: Optional[str] = None) -> int:
223 """Delete tasks older than the specified datetime."""
224 return 0
226 def get_outdated_tasks(
227 self, older_than: datetime, func_name: Optional[str] = None
228 ) -> list[tuple[str, str, str]]:
229 """Retrieve tasks older than the specified datetime (Preview for prune)."""
230 return []
232 def get_blob_refs(self) -> Optional[set[str]]:
233 """Retrieve all 'result_value' entries that point to external storage."""
234 return None
236 def delete_all(self, func_name: Optional[str] = None) -> int:
237 """Delete all tasks, optionally filtered by function name."""
238 return 0
240 def get_keys_start_with(self, prefix: str) -> list[str]:
241 """Retrieve cache keys that start with the given prefix."""
242 return []
244 def get_history(self, limit: int = 1000) -> "pd.DataFrame":
245 """Get task history. Returns an empty DataFrame by default."""
246 try:
247 import pandas as pd
249 return pd.DataFrame()
250 except ImportError:
251 raise ImportError("Pandas is required for this feature.")
254class SQLiteTaskDB(TaskDBCore, Flushable, Shutdownable, Maintenable):
255 """
256 Default implementation using SQLite.
257 """
259 def __init__(self, db_path: str | Path | None = None, timeout: float = 30.0):
260 self.db_path = (
261 Path(db_path).resolve() if db_path else Path(f".beautyspot/{hash(self)}.db")
262 )
263 self._ensure_cache_dir(self.db_path.parent)
264 self.timeout = timeout
265 self._local = threading.local()
266 self._write_queue: queue.Queue[object] = queue.Queue()
267 self._shutdown_lock = threading.Lock()
268 self._shutdown_requested = False
269 self._writer_ready = threading.Event()
270 self._writer_error: Exception | None = None
271 # 読み取り専用スレッドローカル接続を追跡し、
272 # shutdown() 時に一括クローズする。WAL チェックポイントの妨げを防ぐ。
273 self._read_wrappers = weakref.WeakSet()
274 self._read_conns_lock = threading.Lock()
275 self._writer_thread = threading.Thread(
276 target=self._writer_loop, daemon=True, name="BeautySpot-SQLiteWriter"
277 )
278 self._writer_thread.start()
279 self._writer_ready.wait()
280 if self._writer_error: 280 ↛ 281line 280 didn't jump to line 281 because the condition on line 280 was never true
281 raise self._writer_error
283 @staticmethod
284 def _ensure_cache_dir(directory: Path) -> None:
285 """
286 データベース格納用の親ディレクトリを作成し、.gitignore を配置する。
287 """
288 directory.mkdir(parents=True, exist_ok=True)
289 gitignore_path = directory / ".gitignore"
290 if not gitignore_path.exists():
291 try:
292 gitignore_path.write_text("*\n")
293 except OSError as e:
294 logging.warning(f"Failed to create .gitignore in {directory}: {e}")
296 @contextmanager
297 def _read_connect(self) -> Iterator[sqlite3.Connection]:
298 """
299 Thread-safe connection context manager for read-only operations.
300 Uses a thread-local pool to reuse connections and reduce overhead.
301 PRAGMA query_only = ON により、誤った書き込みを接続レベルで防止する。
303 新規接続を _read_wrappers に登録し、
304 shutdown() 時の一括クローズで WAL チェックポイント妨害を防ぐ。
305 また、_ReadConnWrapper によってスレッド終了時に接続がクローズされ、メモリリークを防止。
306 """
307 if self._shutdown_requested: 307 ↛ 308line 307 didn't jump to line 308 because the condition on line 307 was never true
308 raise RuntimeError("SQLiteTaskDB is shutting down.")
310 wrapper = getattr(self._local, "read_conn_wrapper", None)
311 if wrapper is None or wrapper._closed:
312 # シャットダウン後に新しい接続がリークするのを防ぐため再チェック。
313 # 最初のチェック通過後に別スレッドが shutdown() を呼び出し、
314 # 全ラッパーをクローズした場合にここに到達する。
315 if self._shutdown_requested: 315 ↛ 316line 315 didn't jump to line 316 because the condition on line 315 was never true
316 raise RuntimeError("SQLiteTaskDB is shutting down.")
317 conn = sqlite3.connect(
318 self.db_path, timeout=self.timeout, check_same_thread=False
319 )
320 try:
321 conn.execute("PRAGMA query_only = ON;")
322 except Exception:
323 conn.close()
324 raise
325 wrapper = _ReadConnWrapper(conn)
326 with self._read_conns_lock:
327 # ロック内で再度チェックし、shutdown() による _read_wrappers.clear() と
328 # 新規追加の間の競合を完全に排除する。
329 if self._shutdown_requested: 329 ↛ 330line 329 didn't jump to line 330 because the condition on line 329 was never true
330 conn.close()
331 raise RuntimeError("SQLiteTaskDB is shutting down.")
332 self._read_wrappers.add(wrapper)
333 self._local.read_conn_wrapper = wrapper
335 with wrapper.lock:
336 if wrapper._closed: 336 ↛ 337line 336 didn't jump to line 337 because the condition on line 336 was never true
337 raise RuntimeError("Database connection was closed")
338 try:
339 yield wrapper.conn
340 except sqlite3.Error:
341 # 接続が壊れた場合等のリカバリ (BUG-3)
342 # 現在の接続を破棄し、次回のアクセス時に新しく作り直す
343 wrapper.close()
344 with self._read_conns_lock:
345 self._read_wrappers.discard(wrapper)
346 self._local.read_conn_wrapper = None
347 raise
349 # クエリ実行中にシャットダウン要求があった場合、自身でクローズする
350 if getattr(wrapper, "_shutdown_requested", False): 350 ↛ 351line 350 didn't jump to line 351 because the condition on line 350 was never true
351 wrapper.close()
352 with self._read_conns_lock:
353 self._read_wrappers.discard(wrapper)
354 self._local.read_conn_wrapper = None
356 def _writer_loop(self) -> None:
357 conn: sqlite3.Connection | None = None
358 try:
359 conn = sqlite3.connect(self.db_path, timeout=self.timeout)
360 conn.execute("PRAGMA journal_mode=WAL;")
361 except Exception as e:
362 self._writer_error = e
363 self._writer_ready.set()
364 return
366 self._writer_ready.set()
367 try:
368 while True:
369 task = self._write_queue.get()
370 if task is _STOP:
371 self._write_queue.task_done()
372 break
373 assert isinstance(task, _WriteTask)
374 if not task.try_start():
375 # CANCELLED 状態 — スキップ
376 task.event.set()
377 self._write_queue.task_done()
378 continue
380 try:
381 task.result = task.fn(conn)
382 conn.commit()
383 except BaseException as e:
384 conn.rollback()
385 task.error = e
386 finally:
387 task.mark_done()
388 task.event.set()
389 self._write_queue.task_done()
390 finally:
391 if conn is not None: 391 ↛ exitline 391 didn't return from function '_writer_loop' because the condition on line 391 was always true
392 conn.close()
394 def _enqueue_write(self, fn: Callable[[sqlite3.Connection], Any]) -> Any:
395 self._writer_ready.wait()
396 if self._writer_error: 396 ↛ 397line 396 didn't jump to line 397 because the condition on line 396 was never true
397 raise RuntimeError(
398 "SQLite writer thread failed to start."
399 ) from self._writer_error
401 with self._shutdown_lock:
402 if self._shutdown_requested: 402 ↛ 403line 402 didn't jump to line 403 because the condition on line 402 was never true
403 raise RuntimeError("SQLiteTaskDB is shutting down.")
404 if not self._writer_thread.is_alive(): 404 ↛ 405line 404 didn't jump to line 405 because the condition on line 404 was never true
405 raise RuntimeError("SQLite writer thread is not running.")
406 task = _WriteTask(fn=fn, event=threading.Event())
407 self._write_queue.put(task)
409 start = time.monotonic()
410 _warned_running = False
411 while not task.event.wait(timeout=0.5):
412 if not self._writer_thread.is_alive(): 412 ↛ 413line 412 didn't jump to line 413 because the condition on line 412 was never true
413 raise RuntimeError("SQLite writer thread stopped unexpectedly.")
414 if self._shutdown_requested: 414 ↛ 415line 414 didn't jump to line 415 because the condition on line 414 was never true
415 raise RuntimeError("SQLiteTaskDB is shutting down.")
416 elapsed = time.monotonic() - start
417 if elapsed > self.timeout: 417 ↛ 411line 417 didn't jump to line 411 because the condition on line 417 was always true
418 if task.try_cancel():
419 # PENDING（未着手）のタスクはキャンセル可能
420 raise TimeoutError(
421 f"SQLite write did not start within {self.timeout}s and was cancelled."
422 )
423 elif not _warned_running: 423 ↛ 411line 423 didn't jump to line 411 because the condition on line 423 was always true
424 # RUNNING（実行中）のタスクはキャンセル不可。
425 # 旧実装では RUNNING でも TimeoutError を送出していたが、
426 # DB への書き込みは継続されるため呼び出し元との整合性が取れなかった。
427 # 修正後は完了まで待ち続け、警告ログのみ出力する。
428 logger.warning(
429 f"SQLite write has been running for over {self.timeout}s. "
430 "The operation cannot be cancelled and will continue until completion."
431 )
432 _warned_running = True
433 # RUNNING 状態: 完了まで待ち続ける（TimeoutError は送出しない）
434 if task.error: 434 ↛ 435line 434 didn't jump to line 435 because the condition on line 434 was never true
435 raise task.error
436 return task.result
438 def shutdown(self, wait: bool = True) -> None:
439 with self._shutdown_lock:
440 if self._shutdown_requested: 440 ↛ 441line 440 didn't jump to line 441 because the condition on line 440 was never true
441 return
442 self._shutdown_requested = True
444 if not self._writer_thread.is_alive(): 444 ↛ 445line 444 didn't jump to line 445 because the condition on line 444 was never true
445 logger.error(
446 "SQLite writer thread is not running; pending writes may be lost."
447 )
448 return
450 if wait:
451 self._write_queue.join()
452 self._write_queue.put(_STOP)
453 if wait:
454 self._writer_thread.join()
456 # 全スレッドのread-only接続を一括クローズ。
457 # スレッドローカル接続が開いたままだと WAL チェックポイントを妨げるため。
458 with self._read_conns_lock:
459 for wrapper in self._read_wrappers:
460 try:
461 wrapper.close(wait=False)
462 except Exception:
463 pass
464 self._read_wrappers.clear()
466 def init_schema(self):
467 # スキーマ初期化および全マイグレーションを Writer Thread の接続で実行する。
468 # _connect() による別コネクション経由の DDL は、Writer Thread が保持する
469 # WAL ライターロックと競合するリスクがあるため、_enqueue_write に委譲する。
470 def _op(conn: sqlite3.Connection) -> None:
471 conn.execute("PRAGMA journal_mode=WAL;")
472 conn.execute("""
473 CREATE TABLE IF NOT EXISTS tasks (
474 cache_key TEXT PRIMARY KEY,
475 func_name TEXT,
476 func_identifier TEXT,
477 input_id TEXT,
478 result_type TEXT,
479 result_value TEXT,
480 result_data BLOB,
481 content_type TEXT,
482 version TEXT,
483 expires_at TIMESTAMP,
484 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
485 )
486 """)
488 # Auto Migration
489 cursor = conn.execute("PRAGMA table_info(tasks)")
490 columns = [row[1] for row in cursor.fetchall()]
492 if "content_type" not in columns: 492 ↛ 493line 492 didn't jump to line 493 because the condition on line 492 was never true
493 try:
494 conn.execute("ALTER TABLE tasks ADD COLUMN content_type TEXT;")
495 except sqlite3.OperationalError as e:
496 if "duplicate column name" not in str(e).lower():
497 raise
498 pass
499 if "version" not in columns: 499 ↛ 500line 499 didn't jump to line 500 because the condition on line 499 was never true
500 try:
501 conn.execute("ALTER TABLE tasks ADD COLUMN version TEXT;")
502 except sqlite3.OperationalError as e:
503 if "duplicate column name" not in str(e).lower():
504 raise
505 pass
506 if "result_data" not in columns: 506 ↛ 507line 506 didn't jump to line 507 because the condition on line 506 was never true
507 try:
508 conn.execute("ALTER TABLE tasks ADD COLUMN result_data BLOB;")
509 except sqlite3.OperationalError as e:
510 if "duplicate column name" not in str(e).lower():
511 raise
512 pass
514 if "func_identifier" not in columns: 514 ↛ 515line 514 didn't jump to line 515 because the condition on line 514 was never true
515 try:
516 conn.execute("ALTER TABLE tasks ADD COLUMN func_identifier TEXT;")
517 except sqlite3.OperationalError as e:
518 if "duplicate column name" not in str(e).lower():
519 raise
520 pass
521 conn.execute(
522 "CREATE INDEX IF NOT EXISTS idx_func_identifier ON tasks(func_identifier);"
523 )
525 if "expires_at" not in columns:
526 try:
527 conn.execute("ALTER TABLE tasks ADD COLUMN expires_at TIMESTAMP;")
528 except sqlite3.OperationalError as e:
529 if "duplicate column name" not in str(e).lower():
530 raise
531 pass
532 conn.execute(
533 "CREATE INDEX IF NOT EXISTS idx_expires_at ON tasks(expires_at);"
534 )
536 self._enqueue_write(_op)
538 def get(
539 self, cache_key: str, *, include_expired: bool = False
540 ) -> Optional[TaskRecord]:
541 with self._read_connect() as conn:
542 # [MOD] Include expires_at in query
543 row = conn.execute(
544 "SELECT result_type, result_value, result_data, expires_at FROM tasks WHERE cache_key=?",
545 (cache_key,),
546 ).fetchone()
548 if row:
549 r_type, r_val, r_data, exp_str = row
551 # [ADD] Lazy Expiration Check (skip when include_expired=True)
552 if exp_str and not include_expired:
553 try:
554 # SQLite returns timestamps as strings usually
555 # Replace space with T for compatibility with Python <= 3.10
556 expires_at = datetime.fromisoformat(exp_str.replace(" ", "T"))
557 # Naive datetimes stored before timezone support are treated as UTC
558 if expires_at.tzinfo is None: 558 ↛ 559line 558 didn't jump to line 559 because the condition on line 558 was never true
559 expires_at = expires_at.replace(tzinfo=timezone.utc)
560 if expires_at < datetime.now(timezone.utc):
561 # Expired -> Treat as Cache Miss
562 # (Physical deletion is deferred to `beautyspot gc`)
563 return None
564 except (ValueError, TypeError):
565 pass # Ignore parsing errors, treat as valid
567 return TaskRecord(
568 result_type=r_type,
569 result_value=r_val,
570 result_data=r_data,
571 expires_at=exp_str,
572 )
573 return None
575 def save(
576 self,
577 cache_key: str,
578 func_name: str,
579 func_identifier: Optional[str],
580 input_id: str,
581 version: Optional[str],
582 result_type: str,
583 content_type: Optional[str],
584 result_value: Optional[str] = None,
585 result_data: Optional[bytes] = None,
586 expires_at: Optional[datetime] = None, # [ADD] Added argument
587 ):
588 def _op(conn: sqlite3.Connection) -> None:
589 effective_identifier = func_identifier or func_name
590 # updated_at を明示的に設定し、expires_at と同じ形式
591 # (_ensure_utc_isoformat) で統一する。
592 # SQLite の DEFAULT CURRENT_TIMESTAMP は秒精度でフォーマットが異なるため、
593 # prune/get_outdated_tasks との比較精度を揃える。
594 now_str = _ensure_utc_isoformat(datetime.now(timezone.utc))
595 conn.execute(
596 """
597 INSERT OR REPLACE INTO tasks
598 (cache_key, func_name, func_identifier, input_id, version, result_type, content_type, result_value, result_data, expires_at, updated_at)
599 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
600 """,
601 (
602 cache_key,
603 func_name,
604 effective_identifier,
605 input_id,
606 version,
607 result_type,
608 content_type,
609 result_value,
610 result_data,
611 _ensure_utc_isoformat(expires_at),
612 now_str,
613 ),
614 )
616 self._enqueue_write(_op)
618 def get_history(self, limit: int = 1000) -> "pd.DataFrame":
619 try:
620 import pandas as pd
621 except ImportError as e:
622 raise ImportError("Pandas is required for this feature.") from e
624 if not os.path.exists(self.db_path): 624 ↛ 625line 624 didn't jump to line 625 because the condition on line 624 was never true
625 return pd.DataFrame()
627 with self._read_connect() as conn:
628 query = """
629 SELECT
630 cache_key, func_name, func_identifier, input_id, version, result_type,
631 content_type, result_value, result_data, updated_at, expires_at
632 FROM tasks
633 ORDER BY updated_at DESC LIMIT ?
634 """
635 return pd.read_sql_query(query, conn, params=[limit])
637 def delete(self, cache_key: str) -> bool:
638 def _op(conn: sqlite3.Connection) -> bool:
639 cursor = conn.execute("DELETE FROM tasks WHERE cache_key=?", (cache_key,))
640 return cursor.rowcount > 0
642 return bool(self._enqueue_write(_op))
644 def delete_all(self, func_name: Optional[str] = None) -> int:
645 def _op(conn: sqlite3.Connection) -> int:
646 if func_name:
647 cursor = conn.execute(
648 "DELETE FROM tasks WHERE func_name = ? OR func_identifier = ?",
649 (func_name, func_name),
650 )
651 else:
652 cursor = conn.execute("DELETE FROM tasks")
653 return cursor.rowcount
655 return int(self._enqueue_write(_op))
657 def prune(self, older_than: datetime, func_name: Optional[str] = None) -> int:
658 cutoff_str = _ensure_utc_isoformat(older_than)
660 def _op(conn: sqlite3.Connection) -> int:
661 if func_name:
662 cursor = conn.execute(
663 "DELETE FROM tasks WHERE updated_at < ? AND (func_name = ? OR func_identifier = ?)",
664 (cutoff_str, func_name, func_name),
665 )
666 else:
667 cursor = conn.execute(
668 "DELETE FROM tasks WHERE updated_at < ?",
669 (cutoff_str,),
670 )
671 return cursor.rowcount
673 return int(self._enqueue_write(_op))
675 def get_outdated_tasks(
676 self, older_than: datetime, func_name: Optional[str] = None
677 ) -> list[tuple[str, str, str]]:
678 cutoff_str = _ensure_utc_isoformat(older_than)
679 if not os.path.exists(self.db_path): 679 ↛ 680line 679 didn't jump to line 680 because the condition on line 679 was never true
680 return []
682 with self._read_connect() as conn:
683 if func_name:
684 cursor = conn.execute(
685 "SELECT cache_key, COALESCE(func_identifier, func_name), updated_at FROM tasks "
686 "WHERE updated_at < ? AND (func_name = ? OR func_identifier = ?)",
687 (cutoff_str, func_name, func_name),
688 )
689 else:
690 cursor = conn.execute(
691 "SELECT cache_key, COALESCE(func_identifier, func_name), updated_at FROM tasks WHERE updated_at < ?",
692 (cutoff_str,),
693 )
694 return [(row[0], row[1], str(row[2])) for row in cursor.fetchall()]
696 def delete_expired(self) -> int:
697 if not os.path.exists(self.db_path): 697 ↛ 698line 697 didn't jump to line 698 because the condition on line 697 was never true
698 return 0
700 # save() と同じ _ensure_utc_isoformat を使い、フォーマットを統一する
701 now_str = _ensure_utc_isoformat(datetime.now(timezone.utc))
703 def _op(conn: sqlite3.Connection) -> int:
704 cursor = conn.execute(
705 "DELETE FROM tasks WHERE expires_at IS NOT NULL AND expires_at < ?",
706 (now_str,),
707 )
708 return cursor.rowcount
710 return int(self._enqueue_write(_op))
712 def get_blob_refs(self) -> Optional[set[str]]:
713 if not os.path.exists(self.db_path): 713 ↛ 714line 713 didn't jump to line 714 because the condition on line 713 was never true
714 return set()
716 with self._read_connect() as conn:
717 cursor = conn.execute(
718 "SELECT result_value FROM tasks WHERE result_type = 'FILE' AND result_value IS NOT NULL"
719 )
720 # Return full location strings for precise matching
721 return {row[0] for row in cursor.fetchall() if row[0]}
723 def get_keys_start_with(self, prefix: str) -> list[str]:
724 if not os.path.exists(self.db_path): 724 ↛ 725line 724 didn't jump to line 725 because the condition on line 724 was never true
725 return []
727 with self._read_connect() as conn:
728 # LIKE ワイルドカード文字をエスケープしてプレフィックス検索
729 escaped = (
730 prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
731 )
732 cursor = conn.execute(
733 "SELECT cache_key FROM tasks WHERE cache_key LIKE ? ESCAPE '\\' LIMIT 50",
734 (f"{escaped}%",),
735 )
736 return [row[0] for row in cursor.fetchall()]
738 @staticmethod
739 def count_tasks(db_path: Path, timeout: float = 5.0) -> int:
740 """
741 Writer Thread を起動せずに tasks テーブルの件数を返す軽量ユーティリティ。
742 CLI の一覧表示など、読み込みのみを目的とした用途向け。
743 エラー時は -1 を返す。
744 """
745 try:
746 conn = sqlite3.connect(str(db_path), timeout=timeout)
747 try:
748 # 読み取り専用ユーティリティに journal_mode=WAL 設定は不要。
749 # query_only=ON との組み合わせで動作が曖昧になる可能性もあるため削除。
750 conn.execute("PRAGMA query_only = ON;")
751 cursor = conn.execute("SELECT COUNT(*) FROM tasks")
752 result = cursor.fetchone()
753 return result[0] if result else 0
754 finally:
755 conn.close()
756 except Exception:
757 return -1
759 def flush(self, timeout: Optional[float] = None) -> bool:
760 """
761 キューに溜まっているすべての書き込み操作が完了するまで待機します。
763 No-op（何もしない）タスクをキューの末尾に挿入し、そのタスクが処理されるまで
764 待機することで、先行するすべてのタスクの完了を保証します。
766 Args:
767 timeout: 待機する最大秒数。タイムアウトした場合は False を返します。
768 None の場合は無期限に待機しますが、ライタースレッドの
769 死活監視ループにより永久ハングは防止されます。
770 """
771 self._writer_ready.wait()
773 # キューをフラッシュするためのダミータスク
774 def _noop_op(conn: sqlite3.Connection) -> None:
775 pass
777 task = _WriteTask(fn=_noop_op, event=threading.Event())
779 # shutdown との TOCTOU を防ぐため、ロック内でチェックと put を原子的に行う
780 with self._shutdown_lock:
781 if self._shutdown_requested or not self._writer_thread.is_alive(): 781 ↛ 782line 781 didn't jump to line 782 because the condition on line 781 was never true
782 return False
783 self._write_queue.put(task)
785 # ライタースレッドの死活を定期確認しながら待機する。
786 # timeout=None で event.wait() を直接呼ぶとスレッド死亡時に永久ハングするため、
787 # ポーリングループで代替する。
788 _POLL = 0.5
789 deadline = (time.monotonic() + timeout) if timeout is not None else None
791 while True:
792 remaining = (
793 max(0.0, deadline - time.monotonic()) if deadline is not None else None
794 )
795 wait_time = _POLL if remaining is None else min(_POLL, remaining)
797 if task.event.wait(timeout=wait_time):
798 return True
800 if not self._writer_thread.is_alive(): 800 ↛ 801line 800 didn't jump to line 801 because the condition on line 800 was never true
801 logger.error(
802 "SQLite writer thread died unexpectedly while waiting for flush."
803 )
804 return False
806 if deadline is not None and time.monotonic() >= deadline:
807 return False