File size: 1,731 Bytes
35676b4 | 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 | import json
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Optional
from storage.metrics_db import DB_PATH
def _init() -> None:
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
with sqlite3.connect(DB_PATH) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS av_cache (
cache_key TEXT PRIMARY KEY,
payload TEXT NOT NULL,
fetched_at TEXT NOT NULL
)
""")
def get(key: str, ttl_hours: int = 24) -> Optional[dict]:
if not DB_PATH.exists():
return None
try:
with sqlite3.connect(DB_PATH) as conn:
row = conn.execute(
"SELECT payload, fetched_at FROM av_cache WHERE cache_key = ?", (key,)
).fetchone()
if row is None:
return None
payload_str, fetched_at_str = row
fetched_at = datetime.fromisoformat(fetched_at_str)
if datetime.now(timezone.utc) - fetched_at > timedelta(hours=ttl_hours):
return None
return json.loads(payload_str)
except Exception:
return None
def set(key: str, payload: dict) -> None:
_init()
try:
with sqlite3.connect(DB_PATH) as conn:
conn.execute(
"""
INSERT INTO av_cache (cache_key, payload, fetched_at)
VALUES (?, ?, ?)
ON CONFLICT(cache_key) DO UPDATE SET
payload=excluded.payload,
fetched_at=excluded.fetched_at
""",
(key, json.dumps(payload), datetime.now(timezone.utc).isoformat()),
)
except Exception:
pass
|