Class SQLiteCache

requests_cache/backends/sqlite.py:34–190

SQLite cache backend. Args: db_path: Database file path use_cache_dir: Store database in a user cache directory (e.g., `~/.cache/http_cache.sqlite`) use_temp: Store database in a temp directory (e.g., ``/tmp/http_cache.sqlite``) use_memory: Store database in memo

class SQLiteCache(BaseCache):
    """SQLite cache backend.

    Args:
        db_path: Database file path
        use_cache_dir: Store database in a user cache directory (e.g., `~/.cache/http_cache.sqlite`)
        use_temp: Store database in a temp directory (e.g., ``/tmp/http_cache.sqlite``)
        use_memory: Store database in memory instead of in a file
        busy_timeout: Timeout in milliseconds for SQLite to wait if a table is locked.
            See `pragma: busy_timeout <https://www.sqlite.org/pragma.html#pragma_busy_timeout>`_
            for details.
        fast_save: Significantly increases cache write performance, but with the possibility of data
            loss. See `pragma: synchronous <https://www.sqlite.org/pragma.html#pragma_synchronous>`_
            for details.
        wal: Use `Write Ahead Logging <https://sqlite.org/wal.html>`_, so readers do not block writers.
        kwargs: Additional keyword arguments for :py:func:`sqlite3.connect`
    """

    def __init__(
        self,
        db_path: StrOrPath = 'http_cache',
        serializer: Optional[SerializerType] = None,
        **kwargs,
    ):
        super().__init__(cache_name=str(db_path), **kwargs)
        # Only override serializer if a non-None value is specified
        skwargs = {'serializer': serializer, **kwargs} if serializer else kwargs
        self.responses: SQLiteDict = SQLiteDict(db_path, table_name='responses', **skwargs)
        self.redirects: SQLiteDict = SQLiteDict(
            db_path,
            table_name='redirects',
            lock=self.responses._lock,
            serializer=None,
            **kwargs,
        )

    @property
    def db_path(self) -> StrOrPath:
        return self.responses.db_path

    def clear(self):
        """Delete all items from the cache. If this fails due to a corrupted cache or other I/O
        error, this will  attempt to delete the cache file and re-initialize.
        """
        try:
            super().clear()
        except Exception:
            logger.exception('Failed to clear cache')
            if isfile(self.responses.db_path):
                unlink(self.responses.db_path)
            self.responses.init_db()
            self.redirects.init_db()

    # A more efficient SQLite implementation of BaseCache.delete
    def delete(
        self,
        *keys: str,
        expired: bool = False,