Spaces:
Sleeping
Sleeping
File size: 4,294 Bytes
8e54894 | 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 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | """
SQLite-based caching layer for nutrition and literature API responses.
Works both locally and on HF Spaces (file-based, no server needed).
"""
import sqlite3
import json
import hashlib
import time
import os
from pathlib import Path
from typing import Optional
# Default cache location: project root or HF Space persistent storage
CACHE_DIR = os.environ.get("CACHE_DIR", str(Path(__file__).parent.parent))
CACHE_DB = os.path.join(CACHE_DIR, "cache.db")
def _get_connection() -> sqlite3.Connection:
"""Get a SQLite connection with WAL mode for concurrent reads."""
conn = sqlite3.connect(CACHE_DB, timeout=10)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
return conn
def init_cache():
"""Create cache tables if they don't exist."""
conn = _get_connection()
try:
conn.executescript("""
CREATE TABLE IF NOT EXISTS nutrition_cache (
query_key TEXT PRIMARY KEY,
response_json TEXT NOT NULL,
created_at REAL NOT NULL,
ttl_days INTEGER DEFAULT 30
);
CREATE TABLE IF NOT EXISTS literature_cache (
query_key TEXT PRIMARY KEY,
response_json TEXT NOT NULL,
created_at REAL NOT NULL,
ttl_days INTEGER DEFAULT 7
);
""")
conn.commit()
finally:
conn.close()
def _make_key(text: str) -> str:
"""Create a normalized cache key from query text."""
normalized = text.strip().lower()
return hashlib.sha256(normalized.encode()).hexdigest()[:32]
def get_cached(table: str, query: str) -> Optional[dict]:
"""
Retrieve a cached response if it exists and hasn't expired.
Args:
table: 'nutrition_cache' or 'literature_cache'
query: the search query string
Returns:
Parsed JSON dict if cache hit and not expired, None otherwise.
"""
init_cache()
key = _make_key(query)
conn = _get_connection()
try:
row = conn.execute(
f"SELECT response_json, created_at, ttl_days FROM {table} WHERE query_key = ?",
(key,),
).fetchone()
if row is None:
return None
response_json, created_at, ttl_days = row
age_days = (time.time() - created_at) / 86400
if age_days > ttl_days:
# Expired, clean up
conn.execute(f"DELETE FROM {table} WHERE query_key = ?", (key,))
conn.commit()
return None
return json.loads(response_json)
finally:
conn.close()
def set_cached(table: str, query: str, data: dict, ttl_days: int = None):
"""
Store a response in the cache.
Args:
table: 'nutrition_cache' or 'literature_cache'
query: the search query string
data: response dict to cache
ttl_days: override default TTL
"""
init_cache()
key = _make_key(query)
ttl = ttl_days or (30 if table == "nutrition_cache" else 7)
conn = _get_connection()
try:
conn.execute(
f"""INSERT OR REPLACE INTO {table}
(query_key, response_json, created_at, ttl_days)
VALUES (?, ?, ?, ?)""",
(key, json.dumps(data), time.time(), ttl),
)
conn.commit()
finally:
conn.close()
def clear_expired():
"""Remove all expired entries from all cache tables."""
init_cache()
conn = _get_connection()
now = time.time()
try:
for table in ["nutrition_cache", "literature_cache"]:
conn.execute(
f"DELETE FROM {table} WHERE (? - created_at) / 86400 > ttl_days",
(now,),
)
conn.commit()
finally:
conn.close()
def cache_stats() -> dict:
"""Return count and size info for monitoring."""
init_cache()
conn = _get_connection()
try:
stats = {}
for table in ["nutrition_cache", "literature_cache"]:
count = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
stats[table] = count
db_size_mb = os.path.getsize(CACHE_DB) / (1024 * 1024)
stats["db_size_mb"] = round(db_size_mb, 2)
return stats
finally:
conn.close()
|