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
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
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
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
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
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
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,))