tessl/pypi-requests-cache
A persistent cache for python requests
76
1.26x
expiration.mddocs/
Cache Policy and Expiration
Cache policy and expiration features provide flexible control over when responses are cached, how long they remain valid, and how cache validation works. This includes support for HTTP Cache-Control headers, URL-specific patterns, conditional requests, and custom expiration logic.
Capabilities
Cache Settings
Central configuration class that controls all aspects of cache behavior.
class CacheSettings:
"""Internal settings class for cache behavior configuration."""
def __init__(
self,
allowable_codes: Iterable[int] = (200,),
allowable_methods: Iterable[str] = ('GET', 'HEAD'),
always_revalidate: bool = False,
cache_control: bool = False,
disabled: bool = False,
expire_after: ExpirationTime = -1,
filter_fn: Optional[FilterCallback] = None,
ignored_parameters: Optional[Iterable[str]] = None,
key_fn: Optional[KeyCallback] = None,
match_headers: Union[Iterable[str], bool] = False,
stale_if_error: Union[bool, int] = False,
stale_while_revalidate: Union[bool, int] = False,
urls_expire_after: Optional[ExpirationPatterns] = None,
**kwargs
):
"""
Configure cache behavior settings.
Parameters:
- allowable_codes: Only cache responses with these status codes
- allowable_methods: Only cache these HTTP methods
- always_revalidate: Always validate cached responses with server
- cache_control: Use HTTP Cache-Control headers for expiration
- disabled: Temporarily disable all caching
- expire_after: Default expiration time for all cached responses
- filter_fn: Custom function to determine what responses to cache
- ignored_parameters: Parameters to exclude from cache keys
- key_fn: Custom function for generating cache keys
- match_headers: Headers to include in cache keys for matching
- stale_if_error: Return stale responses when new requests fail
- stale_while_revalidate: Return stale responses while refreshing in background
- urls_expire_after: URL-specific expiration patterns
"""
@classmethod
def from_kwargs(cls, **kwargs) -> 'CacheSettings':
"""Create settings instance from keyword arguments."""Expiration Time Functions
Functions for converting various expiration time formats into standardized values.
def get_expiration_datetime(expire_after: ExpirationTime) -> Optional[datetime]:
"""
Convert expiration value to absolute datetime.
Parameters:
- expire_after: Expiration time in various formats
Returns:
Absolute expiration datetime or None for no expiration
Accepts:
- int/float: seconds from now
- str: ISO datetime or relative time ('1 hour', '30 minutes')
- datetime: absolute expiration time
- timedelta: relative time from now
- None: no expiration
- -1: never expire (NEVER_EXPIRE)
- 0: expire immediately (EXPIRE_IMMEDIATELY)
"""
def get_expiration_seconds(expire_after: ExpirationTime) -> Optional[float]:
"""
Convert expiration value to seconds from now.
Parameters:
- expire_after: Expiration time in various formats
Returns:
Seconds until expiration or None for no expiration
"""
def get_url_expiration(
url: str,
urls_expire_after: ExpirationPatterns
) -> ExpirationTime:
"""
Get URL-specific expiration time from pattern matching.
Parameters:
- url: Request URL to match
- urls_expire_after: Dict mapping URL patterns to expiration times
Returns:
Expiration time for matching pattern or None if no match
Pattern matching supports:
- Glob patterns: '*.example.com/api/*'
- Regex patterns: compiled regex objects
- Exact URLs: 'https://api.example.com/data'
"""
def add_tzinfo(dt: datetime, timezone: Optional[tzinfo] = None) -> datetime:
"""Add timezone info to naive datetime (defaults to UTC)."""
def utcnow() -> datetime:
"""Get current UTC time with timezone info."""Usage Examples
Basic expiration configuration:
from requests_cache import CachedSession
from datetime import datetime, timedelta
# Simple numeric expiration (seconds)
session = CachedSession('cache', expire_after=3600) # 1 hour
# Using timedelta objects
session = CachedSession('cache', expire_after=timedelta(hours=2))
# Using datetime objects (absolute expiration)
expire_time = datetime.now() + timedelta(days=1)
session = CachedSession('cache', expire_after=expire_time)
# Using string formats
session = CachedSession('cache', expire_after='1 hour')
session = CachedSession('cache', expire_after='30 minutes')
session = CachedSession('cache', expire_after='2023-12-31T23:59:59')URL-specific expiration patterns:
from requests_cache import CachedSession
session = CachedSession(
'cache',
expire_after=3600, # Default: 1 hour
urls_expire_after={
# Fast-changing APIs: 5 minutes
'*.fastapi.com/data': 300,
'https://api.realtime.com/*': '5 minutes',
# Slow-changing data: 1 day
'*.static.com/*': timedelta(days=1),
# Different expiration for different endpoints
'https://api.example.com/user/*': timedelta(hours=1),
'https://api.example.com/posts/*': timedelta(minutes=30),
# Never expire certain responses
'https://api.example.com/constants': -1,
# Using regex patterns
re.compile(r'.*\.example\.com/v\d+/data'): '1 hour'
}
)Cache Actions
Class that translates cache settings and HTTP headers into specific cache actions for each request.
class CacheActions:
"""Translates settings and headers into cache actions for requests."""
@classmethod
def from_request(
cls,
cache_key: str,
request: AnyPreparedRequest,
settings: CacheSettings
) -> 'CacheActions':
"""Create cache actions based on request and settings."""
def update_from_cached_response(
self,
cached_response: Optional[CachedResponse],
create_key_fn: Callable,
**kwargs
) -> None:
"""Update actions based on cached response state."""
def update_from_response(self, response: AnyResponse) -> None:
"""Update actions based on new response headers."""
@property
def cache_key(self) -> str:
"""Cache key for this request."""
@property
def error_504(self) -> bool:
"""Return 504 error instead of making request."""
@property
def expire_after(self) -> ExpirationTime:
"""Computed expiration time for this request."""
@property
def send_request(self) -> bool:
"""Send new HTTP request."""
@property
def resend_request(self) -> bool:
"""Resend request to refresh stale cached response."""
@property
def resend_async(self) -> bool:
"""Resend request asynchronously while using stale response."""
@property
def skip_read(self) -> bool:
"""Skip reading from cache."""
@property
def skip_write(self) -> bool:
"""Skip writing response to cache."""HTTP Cache Directives
Parser for HTTP Cache-Control and related headers that affect caching behavior.
class CacheDirectives:
"""Parses and stores HTTP cache control directives."""
def __init__(self, headers: Mapping[str, str]):
"""
Parse cache directives from HTTP headers.
Parameters:
- headers: HTTP response headers
Parsed directives include:
- Cache-Control header values
- Expires header
- ETag header
- Last-Modified header
"""
# Cache-Control directive properties
@property
def expires(self) -> Optional[datetime]:
"""Expiration time from Expires header."""
@property
def immutable(self) -> bool:
"""immutable directive."""
@property
def max_age(self) -> Optional[int]:
"""max-age directive value in seconds."""
@property
def max_stale(self) -> Optional[int]:
"""max-stale directive value in seconds."""
@property
def min_fresh(self) -> Optional[int]:
"""min-fresh directive value in seconds."""
@property
def must_revalidate(self) -> bool:
"""must-revalidate directive."""
@property
def no_cache(self) -> bool:
"""no-cache directive."""
@property
def no_store(self) -> bool:
"""no-store directive."""
@property
def only_if_cached(self) -> bool:
"""only-if-cached directive."""
@property
def stale_if_error(self) -> Optional[int]:
"""stale-if-error directive value in seconds."""
@property
def stale_while_revalidate(self) -> Optional[int]:
"""stale-while-revalidate directive value in seconds."""
# Validation headers
@property
def etag(self) -> Optional[str]:
"""ETag header value."""
@property
def last_modified(self) -> Optional[str]:
"""Last-Modified header value."""
def set_request_headers(
headers: Optional[MutableMapping[str, str]],
expire_after: ExpirationTime = None,
only_if_cached: bool = False,
refresh: bool = False,
force_refresh: bool = False
) -> MutableMapping[str, str]:
"""
Convert request parameters to appropriate HTTP headers.
Parameters:
- headers: Existing request headers
- expire_after: Override expiration for this request
- only_if_cached: Add Cache-Control: only-if-cached
- refresh: Add Cache-Control: max-age=0
- force_refresh: Add Cache-Control: no-cache
Returns:
Updated headers dict
"""Usage Examples
HTTP cache control integration:
from requests_cache import CachedSession
# Enable HTTP cache control header processing
session = CachedSession(
'cache',
cache_control=True, # Respect Cache-Control headers
expire_after=3600 # Fallback expiration
)
# Server response with Cache-Control: max-age=1800
# Will be cached for 1800 seconds regardless of expire_after setting
response = session.get('https://api.example.com/data')
# Server response with Cache-Control: no-cache
# Will not be cached regardless of settings
response = session.get('https://api.example.com/nocache')Per-request cache control:
# Force refresh (ignore cached version)
response = session.get(
'https://api.example.com/data',
force_refresh=True
)
# Soft refresh (revalidate with server)
response = session.get(
'https://api.example.com/data',
refresh=True
)
# Only return if cached (return 504 if not cached)
response = session.get(
'https://api.example.com/data',
only_if_cached=True
)
# Override expiration for this request
response = session.get(
'https://api.example.com/data',
expire_after=300 # 5 minutes
)Constants and Defaults
Predefined constants for common expiration values and default settings.
# Expiration constants
DO_NOT_CACHE: int # Special value to disable caching for specific responses
EXPIRE_IMMEDIATELY: int = 0 # Expire immediately
NEVER_EXPIRE: int = -1 # Never expire
# Default settings
DEFAULT_CACHE_NAME: str = 'http_cache'
DEFAULT_METHODS: Tuple[str, ...] = ('GET', 'HEAD')
DEFAULT_STATUS_CODES: Tuple[int, ...] = (200,)
DEFAULT_IGNORED_PARAMS: Tuple[str, ...] = (
'Authorization',
'X-API-KEY',
'access_token',
'api_key'
)Usage Examples
Using expiration constants:
from requests_cache import CachedSession, NEVER_EXPIRE, EXPIRE_IMMEDIATELY
session = CachedSession(
'cache',
expire_after=NEVER_EXPIRE, # Never expire by default
urls_expire_after={
'*.temp.com/*': EXPIRE_IMMEDIATELY, # Always fetch fresh
'*.static.com/*': NEVER_EXPIRE, # Never expire
}
)Custom filtering and key generation:
from requests_cache import CachedSession
def should_cache_response(response):
"""Only cache successful responses from trusted domains."""
if response.status_code != 200:
return False
if 'trusted.com' not in response.url:
return False
return True
def custom_cache_key(*args, **kwargs):
"""Generate cache key that ignores user-specific parameters."""
# Custom key generation logic
return f"custom_key_{hash(args)}"
session = CachedSession(
'cache',
filter_fn=should_cache_response,
key_fn=custom_cache_key,
ignored_parameters=['user_id', 'session_token']
)Types
# Expiration types
ExpirationTime = Union[None, int, float, str, datetime, timedelta]
ExpirationPattern = Union[str, Pattern] # Glob string or compiled regex
ExpirationPatterns = Dict[ExpirationPattern, ExpirationTime]
# Callback types
FilterCallback = Callable[[Response], bool]
KeyCallback = Callable[..., str]
# Header type
HeaderDict = MutableMapping[str, str]Install with Tessl CLI
npx tessl i tessl/pypi-requests-cacheevals
scenario-1
scenario-2
scenario-3
scenario-4
scenario-5
scenario-6
scenario-7
scenario-8
scenario-9
scenario-10