niquests-cache¶

GitHub commits since latest release (by SemVer including pre-releases)

Cached niquests sessions with pluggable storage backends.

Example usage¶

The cached_session helper returns a CachedSession or AsyncCachedSession whose cache database is platformdirs.user_cache_path(app_name, appauthor=False) / 'http.sqlite'. If you omit app_name, niquests-cache is used. Only successful GET and HEAD responses are stored; the default TTL is 10 minutes (expire_after= on the helper, or per request—see below).

Sync helper (default application name and TTL):

from niquests_cache import cached_session

session = cached_session()
response = session.get('https://httpbin.org/get')
response.raise_for_status()

Custom app_name for platformdirs.user_cache_path:

from niquests_cache import cached_session

session = cached_session(app_name='my-application')
response = session.get('https://httpbin.org/get')
response.raise_for_status()

Plain niquests session with no caching:

from niquests_cache import cached_session

session = cached_session(no_cache=True)

Construct CachedSession when you need an explicit cache name or backend:

from pathlib import Path

from niquests_cache import CachedSession

with CachedSession(cache_name=Path('.cache') / 'http') as session:
    response = session.get('https://httpbin.org/get')
    response.raise_for_status()

Async helper (async context manager):

import asyncio
from datetime import timedelta

from niquests_cache import cached_session

async def main() -> None:
    async with cached_session(aio=True, expire_after=timedelta(minutes=30)) as session:
        response = await session.get('https://httpbin.org/get')
        response.raise_for_status()

asyncio.run(main())

Or construct AsyncCachedSession directly:

import asyncio
from pathlib import Path

from niquests_cache import AsyncCachedSession

async def main() -> None:
    async with AsyncCachedSession(cache_name=Path('.cache') / 'http') as session:
        response = await session.get('https://httpbin.org/get')
        response.raise_for_status()

asyncio.run(main())

Pick a non-default backend by alias or instance:

from niquests_cache import CachedSession
from niquests_cache.backends import FileCache, MemoryBackend

# Built-in aliases: 'sqlite' (default), 'filesystem', 'memory'.
session = CachedSession(backend='filesystem', cache_name='./fs-cache')

# Or pass a backend instance directly.
session = CachedSession(backend=MemoryBackend())

# FileCache with a non-default serialiser:
session = CachedSession(backend=FileCache('./fs-cache', serializer='pickle'))

To bypass the cache read for one request, pass force_refresh=True to request(); to return a synthesised 504 when no cached entry exists, pass only_if_cached=True.

API reference¶

Cached niquests sessions matching requests_cache.CachedSession’s API.

class niquests_cache.session.AsyncCachedSession(cache_name: StrPath = 'http_cache', backend: BaseBackend | BackendAlias | None = None, *, serializer: Serializer | str | None = None, expire_after: ExpireAfter = -1, urls_expire_after: Mapping[str | re.Pattern[str], ExpireAfter] | None = None, cache_control: bool = False, content_root_key: str | None = None, allowable_codes: Iterable[int] = (200,), allowable_methods: Iterable[str] = ('GET', 'HEAD'), always_revalidate: bool = False, ignored_parameters: Iterable[str] = ('Authorization', 'If-Modified-Since', 'If-None-Match', 'X-API-KEY', 'access_token', 'api_key'), match_headers: bool | Iterable[str] = False, filter_fn: collections.abc.Callable[..., bool] | None = None, key_fn: collections.abc.Callable[..., str] | None = None, read_only: bool = False, stale_if_error: bool | int = False, autoclose: bool = True, **kwargs: Any)¶

Asynchronous cached niquests session.

cache_disabled() → AsyncIterator[None]¶

Temporarily disable caching for the duration of the async with block.

Yields:¶: None – Control returns to the caller while caching is disabled.

async request(method: str, url: str, *args: Any, headers: Mapping[str, str] | None = None, expire_after: ExpireAfter = None, only_if_cached: bool = False, refresh: bool = False, force_refresh: bool = False, **kwargs: Any) → niquests.Response¶

Send an async request, returning a cached response when available.

Parameters:¶

method : str¶: The HTTP method.
url : str¶: The URL.
*args : Any¶: Positional arguments forwarded to the parent.
headers : Mapping[str, str] | None¶: Per-request headers.
expire_after : ExpireAfter¶: Per-request override of the cache expiry.
only_if_cached : bool¶: If True, return a synthesised 504 when the entry is missing.
refresh : bool¶: Currently treated like force_refresh; reserved for revalidation.
force_refresh : bool¶: If True, bypass the cache read and replace the stored entry.
**kwargs : Any¶: Additional keyword arguments forwarded to the parent.

Returns:¶

The HTTP response.

Return type:¶

niquests.Response

class niquests_cache.session.CacheMixin(cache_name: StrPath = 'http_cache', backend: BaseBackend | BackendAlias | None = None, *, serializer: Serializer | str | None = None, expire_after: ExpireAfter = -1, urls_expire_after: Mapping[str | re.Pattern[str], ExpireAfter] | None = None, cache_control: bool = False, content_root_key: str | None = None, allowable_codes: Iterable[int] = (200,), allowable_methods: Iterable[str] = ('GET', 'HEAD'), always_revalidate: bool = False, ignored_parameters: Iterable[str] = ('Authorization', 'If-Modified-Since', 'If-None-Match', 'X-API-KEY', 'access_token', 'api_key'), match_headers: bool | Iterable[str] = False, filter_fn: collections.abc.Callable[..., bool] | None = None, key_fn: collections.abc.Callable[..., str] | None = None, read_only: bool = False, stale_if_error: bool | int = False, autoclose: bool = True, **kwargs: Any)¶

Shared state and helpers between sync and async cached sessions.

property backend : BaseBackend¶: Alias for cache.

property cache : BaseBackend¶: The storage backend in use.

settings : CacheSettings¶: Mutable cache settings for this session.

class niquests_cache.session.CachedSession(cache_name: StrPath = 'http_cache', backend: BaseBackend | BackendAlias | None = None, *, serializer: Serializer | str | None = None, expire_after: ExpireAfter = -1, urls_expire_after: Mapping[str | re.Pattern[str], ExpireAfter] | None = None, cache_control: bool = False, content_root_key: str | None = None, allowable_codes: Iterable[int] = (200,), allowable_methods: Iterable[str] = ('GET', 'HEAD'), always_revalidate: bool = False, ignored_parameters: Iterable[str] = ('Authorization', 'If-Modified-Since', 'If-None-Match', 'X-API-KEY', 'access_token', 'api_key'), match_headers: bool | Iterable[str] = False, filter_fn: collections.abc.Callable[..., bool] | None = None, key_fn: collections.abc.Callable[..., str] | None = None, read_only: bool = False, stale_if_error: bool | int = False, autoclose: bool = True, **kwargs: Any)¶

Synchronous cached niquests session.

cache_disabled() → Iterator[None]¶

Temporarily disable caching for the duration of the with block.

Yields:¶: None – Control returns to the caller while caching is disabled.

request(method: str, url: str, *args: Any, headers: Mapping[str, str] | None = None, expire_after: ExpireAfter = None, only_if_cached: bool = False, refresh: bool = False, force_refresh: bool = False, **kwargs: Any) → niquests.Response¶

Send a request, returning a cached response when one is available.

Parameters:¶

method : str¶: The HTTP method.
url : str¶: The URL.
*args : Any¶: Positional arguments forwarded to the parent.
headers : Mapping[str, str] | None¶: Per-request headers.
expire_after : ExpireAfter¶: Per-request override of the cache expiry.
only_if_cached : bool¶: If True, return a synthesised 504 when the entry is missing.
refresh : bool¶: Currently treated like force_refresh; reserved for revalidation.
force_refresh : bool¶: If True, bypass the cache read and replace the stored entry.
**kwargs : Any¶: Additional keyword arguments forwarded to the parent.

Returns:¶

The HTTP response.

Return type:¶

niquests.Response

niquests_cache.session.cached_session(*, aio: False = False, no_cache: True, app_name: str | None = None, backend: BaseBackend | BackendAlias | None = None, expire_after: timedelta = datetime.timedelta(seconds=600)) → Session¶

niquests_cache.session.cached_session(*, aio: True, no_cache: True, app_name: str | None = None, backend: BaseBackend | BackendAlias | None = None, expire_after: timedelta = datetime.timedelta(seconds=600)) → AsyncSession

niquests_cache.session.cached_session(*, aio: True, no_cache: False = False, app_name: str | None = None, backend: BaseBackend | BackendAlias | None = None, expire_after: timedelta = datetime.timedelta(seconds=600)) → AsyncCachedSession

niquests_cache.session.cached_session(*, aio: False = False, no_cache: False = False, app_name: str | None = None, backend: BaseBackend | BackendAlias | None = None, expire_after: timedelta = datetime.timedelta(seconds=600)) → CachedSession

niquests_cache.session.cached_session(*, aio: False = False, no_cache: bool = False, app_name: str | None = None, backend: BaseBackend | BackendAlias | None = None, expire_after: timedelta = datetime.timedelta(seconds=600)) → Session

niquests_cache.session.cached_session(*, aio: True, no_cache: bool = False, app_name: str | None = None, backend: BaseBackend | BackendAlias | None = None, expire_after: timedelta = datetime.timedelta(seconds=600)) → AsyncSession

Get a niquests session, optionally with SQLite-backed caching.

The default cache database lives at <user cache path>/http.sqlite.

Parameters:¶

aio : bool¶: If True, return an async session; otherwise a synchronous session.
no_cache : bool¶: If True, return a plain session without caching.
app_name : str | None¶: First argument to platformdirs.user_cache_path() for the cache root. If None, uses niquests-cache.
backend : BaseBackend | BackendAlias | None¶: Storage backend. May be a BaseBackend instance, an alias ('sqlite', 'filesystem', 'memory'), or None (default) to use SQLite at the user-cache path derived from app_name.
expire_after : timedelta¶: Cache expiry duration (ignored when no_cache is True).

Returns:¶

A plain Session or AsyncSession when no_cache is True (depending on aio); otherwise a CachedSession or AsyncCachedSession. Use an async context manager when aio is True.

Return type:¶

CachedSession | AsyncCachedSession | niquests.Session | niquests.AsyncSession

Abstract base class for cache backends.

class niquests_cache.backends.base.BaseBackend¶

Abstract storage backend used by CachedSession.

async aget(key: str) → CacheEntry | None¶

Async get(). Default implementation calls the sync version.

Parameters:¶

key : str¶: The cache key.

Returns:¶

The stored entry, or None if the key is unknown.

Return type:¶

CacheEntry | None

async aset(key: str, entry: CacheEntry) → None¶

Async set(). Default implementation calls the sync version.

Parameters:¶

key : str¶: The cache key.
entry : CacheEntry¶: The entry to store.

abstractmethod get(key: str) → CacheEntry | None¶

Look up a cached entry.

Parameters:¶

key : str¶: The cache key.

Returns:¶

The stored entry, or None if the key is unknown.

Return type:¶

CacheEntry | None

abstractmethod set(key: str, entry: CacheEntry) → None¶

Persist a cache entry.

Parameters:¶

key : str¶: The cache key.
entry : CacheEntry¶: The entry to store.

Filesystem cache backend matching the requests_cache.FileCache public API.

class niquests_cache.backends.file.FileCache(cache_name: StrPath = 'http_cache', *, use_temp: bool = False, use_cache_dir: bool = False, extension: str = '', lock: AbstractContextManager[Any] | None = None, serializer: Serializer | str | None = None, **kwargs: Any)¶

Cache responses as serialised files in a directory.

async aget(key: str) → CacheEntry | None¶

Async get() using anyio for non-blocking I/O.

Parameters:¶

key : str¶: The cache key.

Returns:¶

The stored entry, or None if the file is missing or unreadable.

Return type:¶

CacheEntry | None

async aset(key: str, entry: CacheEntry) → None¶

Async set() using anyio for non-blocking I/O.

Parameters:¶

key : str¶: The cache key.
entry : CacheEntry¶: The entry to store.

property cache_dir : Path¶: Filesystem directory holding cache files.

get(key: str) → CacheEntry | None¶

Look up a cached entry.

Parameters:¶

key : str¶: The cache key.

Returns:¶

The stored entry, or None if the file is missing or unreadable.

Return type:¶

CacheEntry | None

property serializer : Serializer¶: The serialiser used to encode entries to bytes.

set(key: str, entry: CacheEntry) → None¶

Persist a cache entry.

Parameters:¶

key : str¶: The cache key.
entry : CacheEntry¶: The entry to store.

In-memory cache backend.

class niquests_cache.backends.memory.MemoryBackend¶

Cache responses in an in-process dictionary.

get(key: str) → CacheEntry | None¶

Look up a cached entry.

Parameters:¶

key : str¶: The cache key.

Returns:¶

A shallow copy of the stored entry, or None if the key is unknown. The copy isolates callers from the backend’s internal state so mutating the returned entry does not silently update the cache.

Return type:¶

CacheEntry | None

set(key: str, entry: CacheEntry) → None¶

Persist a cache entry.

Parameters:¶

key : str¶: The cache key.
entry : CacheEntry¶: The entry to store.

SQLite cache backend storing entries in typed columns.

class niquests_cache.backends.sqlite.SQLiteBackend(database: StrPath = ':memory:', *, table_name: str = 'niquests_cache')¶

Cache responses in a SQLite database with typed columns.

async aget(key: str) → CacheEntry | None¶

Async get() using aiosqlite.

Parameters:¶

key : str¶: The cache key.

Returns:¶

The stored entry, or None if not present.

Return type:¶

CacheEntry | None

async aset(key: str, entry: CacheEntry) → None¶

Async set() using aiosqlite.

Parameters:¶

key : str¶: The cache key.
entry : CacheEntry¶: The entry to store.

close() → None¶: Close the underlying SQLite connection.

property database : StrPath¶: SQLite database path (or ':memory:').

get(key: str) → CacheEntry | None¶

Look up a cached entry.

Parameters:¶

key : str¶: The cache key.

Returns:¶

The stored entry, or None if not present.

Return type:¶

CacheEntry | None

set(key: str, entry: CacheEntry) → None¶

Persist a cache entry.

Parameters:¶

key : str¶: The cache key.
entry : CacheEntry¶: The entry to store.

Built-in serialisers for cache entries.

class niquests_cache.serializers.JSONSerializer¶

Encodes CacheEntry as UTF-8 JSON. Binary content is base64-encoded.

dumps(entry: CacheEntry) → bytes¶

Serialise entry to JSON-encoded bytes.

Parameters:¶

entry : CacheEntry¶: The cache entry to serialise.

Returns:¶

The encoded payload.

Return type:¶

bytes

loads(data: bytes) → CacheEntry¶

Deserialise a JSON-encoded payload.

Parameters:¶

data : bytes¶: The serialised payload.

Returns:¶

The decoded cache entry.

Return type:¶

CacheEntry

class niquests_cache.serializers.PickleSerializer¶

Encodes CacheEntry using pickle.

dumps(entry: CacheEntry) → bytes¶

Serialise entry using pickle.dumps().

Parameters:¶

entry : CacheEntry¶: The cache entry to serialise.

Returns:¶

The pickled payload.

Return type:¶

bytes

loads(data: bytes) → CacheEntry¶

Deserialise a pickled payload.

Parameters:¶

data : bytes¶: The serialised payload.

Returns:¶

The decoded cache entry.

Return type:¶

CacheEntry

niquests_cache.serializers.resolve_serializer(value: str | Serializer | None) → Serializer¶

Resolve a serializer argument.

Parameters:¶

value : str | Serializer | None¶: Built-in alias ('json' or 'pickle'), an object exposing dumps/loads, or None for the default JSON serialiser.

Returns:¶

The resolved serialiser.

Return type:¶

Serializer

Raises:¶

TypeError – If value is an unknown alias or lacks dumps/loads methods.

Cache settings dataclass for CachedSession.

class niquests_cache.settings.CacheSettings(allowable_codes: tuple[int, ...] = (200,), allowable_methods: tuple[str, ...] = ('GET', 'HEAD'), always_revalidate: bool = False, autoclose: bool = True, cache_control: bool = False, content_root_key: str | None = None, disabled: bool = False, expire_after: ExpireAfter = -1, filter_fn: collections.abc.Callable[..., bool] | None = None, ignored_parameters: tuple[str, ...] = ('Authorization', 'X-API-KEY', 'access_token', 'api_key'), key_fn: collections.abc.Callable[..., str] | None = None, match_headers: bool | tuple[str, ...] = False, read_only: bool = False, serializer: Serializer | None = None, stale_if_error: bool | int = False, urls_expire_after: Mapping[str | re.Pattern[str], ExpireAfter] | None = None)¶

Mutable settings controlling CachedSession behaviour.

allowable_codes : tuple[int, ...] = (200,)¶: HTTP status codes whose responses may be cached.

allowable_methods : tuple[str, ...] = ('GET', 'HEAD')¶: HTTP methods whose responses may be cached.

always_revalidate : bool = False¶: Stored for parity with requests_cache; not used.

autoclose : bool = True¶: Stored for parity with requests_cache; not used.

cache_control : bool = False¶: Honour Cache-Control response headers (currently stored, not enforced).

content_root_key : str | None = None¶: Stored for parity with requests_cache; not used.

disabled : bool = False¶: Internal flag toggled by CachedSession.cache_disabled().

expire_after : ExpireAfter = -1¶: Default TTL; -1 or None means never expire.

filter_fn : Callable[..., bool] | None = None¶: Predicate deciding whether a response should be cached.

ignored_parameters : tuple[str, ...] = ('Authorization', 'X-API-KEY', 'access_token', 'api_key')¶: Header / query-parameter names removed before computing the cache key.

key_fn : Callable[..., str] | None = None¶: Optional callable producing a custom cache key.

match_headers : bool | tuple[str, ...] = False¶

only those names.

Type:¶: False ignores headers; True includes all; iterable

read_only : bool = False¶: If True, never write to the cache.

serializer : Serializer | None = None¶: Resolved serialiser used by the session to encode and decode entries.

stale_if_error : bool | int = False¶: Stored for parity with requests_cache; not used.

urls_expire_after : Mapping[str | re.Pattern[str], ExpireAfter] | None = None¶: Per-URL-pattern TTL overrides.

Typing helpers for niquests_cache.

niquests_cache.typing.BackendAlias¶

Built-in backend alias names accepted by CachedSession.

alias of Literal[‘sqlite’, ‘memory’, ‘filesystem’]

class niquests_cache.typing.CacheEntry¶

A cached HTTP response entry.

content : bytes¶: Response body as raw bytes.

encoding : str¶: Response encoding.

headers : dict[str, str]¶: Response headers.

status_code : int¶: HTTP status code.

ts : float¶: Unix timestamp the entry was written.

url : str¶: Final response URL.

niquests_cache.typing.ExpireAfter : TypeAlias = int | float | str | datetime.datetime | datetime.timedelta | None¶: Accepted types for the expire_after parameter.