sift_py.grpc.cache

Utilities for controlling gRPC response caching.

This module provides helper functions and constants for working with the gRPC caching interceptor. Use these utilities to control caching behavior on a per-request basis via metadata.

Example

from sift_py.grpc.cache import with_cache, with_force_refresh, ignore_cache

Enable caching for a request

metadata = with_cache() response = stub.GetData(request, metadata=metadata)

Force refresh (bypass cache and store fresh result)

metadata = with_force_refresh() response = stub.GetData(request, metadata=metadata)

Ignore cache without clearing

metadata = ignore_cache() response = stub.GetData(request, metadata=metadata)

CLASS	DESCRIPTION
CacheSettings	Resolved cache metadata from gRPC request.
GrpcCache	Subclass of diskcache.Cache for gRPC response caching.

FUNCTION	DESCRIPTION
with_cache	Enable caching for a gRPC request.
with_force_refresh	Force refresh the cache for a gRPC request.

ATTRIBUTE	DESCRIPTION
METADATA_CACHE_TTL
METADATA_FORCE_REFRESH
METADATA_USE_CACHE
logger

METADATA_CACHE_TTL module-attribute

METADATA_CACHE_TTL = 'cache-ttl'

METADATA_FORCE_REFRESH module-attribute

METADATA_FORCE_REFRESH = 'force-refresh'

METADATA_USE_CACHE module-attribute

METADATA_USE_CACHE = 'use-cache'

logger module-attribute

logger = getLogger(__name__)

CacheSettings

Bases: NamedTuple

Resolved cache metadata from gRPC request.

ATTRIBUTE	DESCRIPTION
custom_ttl	TYPE: float | None
force_refresh	TYPE: bool
use_cache	TYPE: bool

custom_ttl instance-attribute

custom_ttl: float | None

force_refresh instance-attribute

force_refresh: bool

use_cache instance-attribute

use_cache: bool

GrpcCache

GrpcCache(config: SiftCacheConfig)

Bases: Cache

Subclass of diskcache.Cache for gRPC response caching.

Initialize the cache from configuration.

PARAMETER	DESCRIPTION
config	Cache configuration with ttl, cache_path, size_limit, clear_on_init. TYPE: SiftCacheConfig

METHOD	DESCRIPTION
key_from_proto_message
resolve_cache_metadata	Extract and resolve cache-related metadata fields.
set_with_default_ttl

ATTRIBUTE	DESCRIPTION
cache_path
default_ttl
size_limit

cache_path instance-attribute

cache_path = Path(config['cache_path'])

default_ttl instance-attribute

default_ttl = config['ttl']

size_limit instance-attribute

size_limit = config['size_limit']

key_from_proto_message staticmethod

key_from_proto_message(
    method_name: str | bytes, request: Message
) -> str

resolve_cache_metadata staticmethod

resolve_cache_metadata(
    metadata: tuple[tuple[str, str], ...] | None,
) -> CacheSettings

Extract and resolve cache-related metadata fields.

PARAMETER	DESCRIPTION
metadata	The gRPC request metadata tuple. TYPE: tuple[tuple[str, str], ...] | None

RETURNS	DESCRIPTION
CacheSettings	CacheMetadata named tuple with resolved cache control fields:
CacheSettings	use_cache: bool - Whether to use caching
CacheSettings	force_refresh: bool - Whether to force refresh
CacheSettings	ignore_cache: bool - Whether to ignore cache
CacheSettings	custom_ttl: int | None - Custom TTL if specified
CacheSettings	should_read: bool - Whether to read from cache
CacheSettings	should_cache: bool - Whether to cache the response

Example

cache_info = cache.resolve_cache_metadata(metadata) if cache_info.should_read: cached = cache.get(key) if cache_info.should_cache: cache.set_with_default_ttl(key, response, expire=cache_info.custom_ttl)

set_with_default_ttl

set_with_default_ttl(
    key: str,
    value: Any,
    expire: float | None = None,
    **kwargs,
) -> bool

with_cache

with_cache(
    ttl: int | None = None,
) -> tuple[tuple[str, str], ...]

Enable caching for a gRPC request.

PARAMETER	DESCRIPTION
ttl	Optional custom TTL in seconds. If not provided, uses the default TTL. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
tuple[tuple[str, str], ...]	Metadata tuple to pass to the gRPC stub method.

Example

metadata = with_cache() response = stub.GetData(request, metadata=metadata)

With custom TTL

metadata = with_cache(ttl=7200) # 2 hours response = stub.GetData(request, metadata=metadata)

with_force_refresh

with_force_refresh(
    ttl: int | None = None,
) -> tuple[tuple[str, str], ...]

Force refresh the cache for a gRPC request.

Bypasses the cache, fetches fresh data from the server, and stores the result.

PARAMETER	DESCRIPTION
ttl	Optional custom TTL in seconds. If not provided, uses the default TTL. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
tuple[tuple[str, str], ...]	Metadata tuple to pass to the gRPC stub method.

Example

metadata = with_force_refresh() response = stub.GetData(request, metadata=metadata)