Class SQLiteCache

app/api/agentops/common/cache.py:77–126

SQLite-backed cache for local development.

                del self.expiry[key]

    class SQLiteCache(BaseDevCache):
        """SQLite-backed cache for local development."""

        def __init__(self):
            self.db_path = os.path.join(os.getcwd(), "cache.db")
            self.conn = sqlite3.connect(self.db_path)
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS cache (
                    key TEXT PRIMARY KEY,
                    value TEXT,
                    expiry INTEGER
                )
            """)
            self.conn.commit()

        def get(self, key: str) -> str | None:
            cursor = self.conn.execute("SELECT value, expiry FROM cache WHERE key = ?", (key,))
            row = cursor.fetchone()
            if row:
                value, expiry = row
                if expiry is not None and time.time() > expiry:
                    self.delete(key)
                    return None
                return value
            return None

        def setex(self, key: str, expiry: int, value: str) -> None:
            expiry_time = int(time.time() + expiry)
            self.conn.execute(
                """
                INSERT OR REPLACE INTO cache (key, value, expiry)
                VALUES (?, ?, ?)
            """,
                (key, value, expiry_time),
            )
            self.conn.commit()

        def expire(self, key: str, expiry: int) -> None:
            expiry_time = int(time.time() + expiry)
            self.conn.execute(
                """
                UPDATE cache SET expiry = ? WHERE key = ?
            """,
                (expiry_time, key),
            )
            self.conn.commit()

        def delete(self, key: str) -> None:
            self.conn.execute("DELETE FROM cache WHERE key = ?", (key,))
            self.conn.commit()

    if os.path.exists("/.dockerenv"):
        logger.info("Using in-memory cache for local development.")