Spaces:
Sleeping
Sleeping
File size: 3,338 Bytes
7d7841f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | import os
import sqlite3
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
_DEFAULT_CACHE_PATH = Path(__file__).resolve().parents[1] / ".cache" / "entity_lookup.sqlite3"
@dataclass(frozen=True)
class CachedEntityLookup:
cache_key: str
status: str
state: str
evidence_level: str
score: int
max_score: int
finding: str
fix: str
checked_at: datetime
def _cache_path() -> Path:
configured = os.getenv("ENTITY_CACHE_PATH")
return Path(configured) if configured else _DEFAULT_CACHE_PATH
def entity_cache_ttl() -> timedelta:
days = int(os.getenv("ENTITY_CACHE_TTL_DAYS", "30"))
return timedelta(days=days)
def _connection() -> sqlite3.Connection:
path = _cache_path()
path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(path)
connection.execute(
"""
CREATE TABLE IF NOT EXISTS entity_lookup_cache (
cache_key TEXT PRIMARY KEY,
status TEXT NOT NULL,
state TEXT NOT NULL,
evidence_level TEXT NOT NULL,
score INTEGER NOT NULL,
max_score INTEGER NOT NULL,
finding TEXT NOT NULL,
fix TEXT NOT NULL,
checked_at TEXT NOT NULL
)
"""
)
return connection
def get_cached_entity_lookup(cache_key: str) -> CachedEntityLookup | None:
connection = _connection()
try:
row = connection.execute(
"""
SELECT cache_key, status, state, evidence_level, score, max_score, finding, fix, checked_at
FROM entity_lookup_cache
WHERE cache_key = ?
""",
(cache_key,),
).fetchone()
finally:
connection.close()
if row is None:
return None
checked_at = datetime.fromisoformat(row[8])
if datetime.now(UTC) - checked_at >= entity_cache_ttl():
return None
return CachedEntityLookup(
cache_key=row[0],
status=row[1],
state=row[2],
evidence_level=row[3],
score=row[4],
max_score=row[5],
finding=row[6],
fix=row[7],
checked_at=checked_at,
)
def set_cached_entity_lookup(
cache_key: str,
status: str,
state: str,
evidence_level: str,
score: int,
max_score: int,
finding: str,
fix: str,
) -> None:
checked_at = datetime.now(UTC).isoformat()
connection = _connection()
try:
connection.execute(
"""
INSERT INTO entity_lookup_cache (
cache_key, status, state, evidence_level, score, max_score, finding, fix, checked_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(cache_key) DO UPDATE SET
status = excluded.status,
state = excluded.state,
evidence_level = excluded.evidence_level,
score = excluded.score,
max_score = excluded.max_score,
finding = excluded.finding,
fix = excluded.fix,
checked_at = excluded.checked_at
""",
(cache_key, status, state, evidence_level, score, max_score, finding, fix, checked_at),
)
connection.commit()
finally:
connection.close()
|