Skip to content

SQLite Cache

SQLiteCache(path: Path | None = None, expiry: int | None = 14)

The SQLiteCache object to cache search results from Comicvine.

PARAMETER DESCRIPTION
path

Path to database.

TYPE: Path | None DEFAULT: None

expiry

How long to keep cache results.

TYPE: int | None DEFAULT: 14

Source code in simyan/cache/sqlite_cache.py
Python
21
22
23
24
25
def __init__(self, path: Path | None = None, expiry: int | None = 14):
    self._db_path = path or (get_cache_root() / "cache.sqlite")
    self._expiry = expiry
    self.initialize()
    self.cleanup()

Functions

cleanup() -> None

Remove all expired entries from the cache database.

Source code in simyan/cache/sqlite_cache.py
Python
 97
 98
 99
100
101
102
103
104
def cleanup(self) -> None:
    """Remove all expired entries from the cache database."""
    if not self._expiry:
        return
    expiry = datetime.now(tz=timezone.utc) - timedelta(days=self._expiry)
    with self._connect() as conn:
        conn.execute("DELETE FROM cache WHERE timestamp < ?;", (expiry.isoformat(),))
        conn.commit()

delete(query: str) -> None

Remove entry from the cache with the provided url.

PARAMETER DESCRIPTION
query

Url string used as key.

TYPE: str

Source code in simyan/cache/sqlite_cache.py
Python
87
88
89
90
91
92
93
94
95
def delete(self, query: str) -> None:
    """Remove entry from the cache with the provided url.

    Args:
      query: Url string used as key.
    """
    with self._connect() as conn:
        conn.execute("DELETE FROM cache WHERE query = ?;", (query,))
        conn.commit()

initialize() -> None

Create the cache table if it doesn't exist.

Source code in simyan/cache/sqlite_cache.py
Python
39
40
41
42
43
44
45
46
47
48
49
50
51
def initialize(self) -> None:
    """Create the cache table if it doesn't exist."""
    with self._connect() as conn:
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS cache (
                query TEXT NOT NULL PRIMARY KEY,
                response TEXT,
                timestamp TIMESTAMP
            );
            """
        )
        conn.commit()

insert(query: str, response: dict[str, Any]) -> None

Insert data into the cache database.

PARAMETER DESCRIPTION
query

Url string used as key.

TYPE: str

response

Response dict from url.

TYPE: dict[str, Any]

Source code in simyan/cache/sqlite_cache.py
Python
73
74
75
76
77
78
79
80
81
82
83
84
85
def insert(self, query: str, response: dict[str, Any]) -> None:
    """Insert data into the cache database.

    Args:
        query: Url string used as key.
        response: Response dict from url.
    """
    with self._connect() as conn:
        conn.execute(
            "INSERT INTO cache (query, response, timestamp) VALUES (?, ?, ?);",
            (query, json.dumps(response), datetime.now(tz=timezone.utc).isoformat()),
        )
        conn.commit()

select(query: str) -> dict[str, Any]

Retrieve data from the cache database.

PARAMETER DESCRIPTION
query

Url string used as key.

TYPE: str

RETURNS DESCRIPTION
dict[str, Any]

Empty dict or select results.
Source code in simyan/cache/sqlite_cache.py
Python
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def select(self, query: str) -> dict[str, Any]:
    """Retrieve data from the cache database.

    Args:
        query: Url string used as key.

    Returns:
        Empty dict or select results.
    """
    with self._connect() as conn:
        if self._expiry:
            expiry = datetime.now(tz=timezone.utc) - timedelta(days=self._expiry)
            row = conn.execute(
                "SELECT * FROM cache WHERE query = ? and timestamp > ?;",
                (query, expiry.isoformat()),
            ).fetchone()
        else:
            row = conn.execute("SELECT * FROM cache WHERE query = ?;", (query,)).fetchone()
        return json.loads(row["response"]) if row else {}