storage/earnings_cache.py

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