Skip to content

SQLite Cache

SQLiteCache

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: timedelta | None DEFAULT: timedelta(days=14)

Source code in simyan/cache/sqlite_cache.py
Python
26
27
28
29
30
def __init__(self, path: Path | None = None, expiry: timedelta | None = timedelta(days=14)):
    self._db_path: Path = path or (get_cache_root() / "cache.sqlite")
    self._expiry: timedelta | None = expiry
    self.initialize()
    self.cleanup()

Functions

cleanup

Remove all expired entries from the cache database.

Source code in simyan/cache/sqlite_cache.py
Python
129
130
131
132
133
134
135
136
137
138
139
def cleanup(self) -> None:
    """Remove all expired entries from the cache database."""
    if not self._expiry:
        return
    query = """
    DELETE FROM queries
    WHERE created_at < ?;
    """
    expiry = datetime.now(tz=timezone.utc) - self._expiry
    with self._connect() as conn:
        conn.execute(query, (expiry.isoformat(),))

delete

Remove entry from the cache with the provided url.

PARAMETER DESCRIPTION
url

The URL used as the cache key.

TYPE: str

Source code in simyan/cache/sqlite_cache.py
Python
116
117
118
119
120
121
122
123
124
125
126
127
def delete(self, url: str) -> None:
    """Remove entry from the cache with the provided url.

    Args:
        url: The URL used as the cache key.
    """
    query = """
    DELETE FROM queries
    WHERE url = ?;
    """
    with self._connect() as conn:
        conn.execute(query, (url,))

initialize

Create the cache table if it doesn't exist.

Source code in simyan/cache/sqlite_cache.py
Python
45
46
47
48
49
50
51
52
53
54
55
def initialize(self) -> None:
    """Create the cache table if it doesn't exist."""
    query = """
    CREATE TABLE IF NOT EXISTS queries (
        url TEXT PRIMARY KEY,
        response TEXT NOT NULL,
        created_at DATETIME NOT NULL
    );
    """
    with self._connect() as conn:
        conn.executescript(query)

insert

Insert data into the cache database.

If a record with the same url already exists it is overwritten (upsert semantics).

PARAMETER DESCRIPTION
url

The URL used as the cache key.

TYPE: str

response

The response body to store.

TYPE: dict[str, Any]

Source code in simyan/cache/sqlite_cache.py
Python
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def insert(self, url: str, response: dict[str, Any]) -> None:
    """Insert data into the cache database.

    If a record with the same *url* already exists it is overwritten (upsert semantics).

    Args:
        url: The URL used as the cache key.
        response: The response body to store.
    """
    query = """
    INSERT INTO queries (url, response, created_at)
    VALUES (?, ?, ?)
    ON CONFLICT (url) DO UPDATE SET
        response = excluded.response,
        created_at = excluded.created_at;
    """
    with self._connect() as conn:
        conn.execute(
            query, (url, json.dumps(response), datetime.now(tz=timezone.utc).isoformat())
        )

select

Retrieve data from the cache database.

PARAMETER DESCRIPTION
url

The URL used as the cache key.

TYPE: str

RETURNS DESCRIPTION
CacheData | None

A CacheData instance if a matching record exists, otherwise None.
Source code in simyan/cache/sqlite_cache.py
Python
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def select(self, url: str) -> CacheData | None:
    """Retrieve data from the cache database.

    Args:
        url: The URL used as the cache key.

    Returns:
        A `CacheData` instance if a matching record exists, otherwise `None`.
    """
    if self._expiry:
        query = """
        SELECT url, response, created_at
        FROM queries
        WHERE url = ? AND created_at > ?;
        """
        expiry = datetime.now(tz=timezone.utc) - self._expiry
        return self._select(query=query, params=(url, expiry.isoformat()))
    query = """
    SELECT url, response, created_at
    FROM queries
    WHERE url = ?;
    """
    return self._select(query=query, params=(url,))