A persistent cache for python requests
76
Session-based caching provides the primary interface for requests-cache, offering drop-in replacement for requests.Session with transparent caching capabilities. This approach gives you full control over cache settings and behavior while maintaining compatibility with existing requests-based code.
The main session class that extends requests.Session with caching features. Provides all standard HTTP methods (GET, POST, PUT, etc.) with automatic caching based on configurable policies.
class CachedSession:
def __init__(
self,
cache_name: StrOrPath = 'http_cache',
backend: Optional[BackendSpecifier] = None,
serializer: Optional[SerializerType] = None,
expire_after: ExpirationTime = -1,
urls_expire_after: Optional[ExpirationPatterns] = None,
cache_control: bool = False,
allowable_codes: Iterable[int] = (200,),
allowable_methods: Iterable[str] = ('GET', 'HEAD'),
always_revalidate: bool = False,
ignored_parameters: Iterable[str] = None,
match_headers: Union[Iterable[str], bool] = False,
filter_fn: Optional[FilterCallback] = None,
key_fn: Optional[KeyCallback] = None,
stale_if_error: Union[bool, int] = False,
**kwargs
):
"""
Initialize cached session with caching features.
Parameters:
- cache_name: Cache path, prefix, or namespace depending on backend
- backend: Backend name ('sqlite', 'redis', 'mongodb', etc.) or instance
- serializer: Serializer name ('pickle', 'json', 'yaml', 'bson') or instance
- expire_after: Default expiration time for cached responses
- urls_expire_after: URL-specific expiration patterns
- cache_control: Use HTTP Cache-Control headers for expiration
- allowable_codes: Only cache responses with these status codes
- allowable_methods: Cache only these HTTP methods
- always_revalidate: Always validate cached responses with server
- ignored_parameters: Parameters to exclude from cache keys
- match_headers: Headers to include in cache keys
- filter_fn: Custom function to determine what responses to cache
- key_fn: Custom function for generating cache keys
- stale_if_error: Return stale responses on errors
"""
def get(self, url: str, params=None, **kwargs) -> AnyResponse: ...
def post(self, url: str, data=None, **kwargs) -> AnyResponse: ...
def put(self, url: str, data=None, **kwargs) -> AnyResponse: ...
def patch(self, url: str, data=None, **kwargs) -> AnyResponse: ...
def delete(self, url: str, **kwargs) -> AnyResponse: ...
def head(self, url: str, **kwargs) -> AnyResponse: ...
def options(self, url: str, **kwargs) -> AnyResponse: ...
def request(
self,
method: str,
url: str,
headers: Optional[MutableMapping[str, str]] = None,
expire_after: ExpirationTime = None,
only_if_cached: bool = False,
refresh: bool = False,
force_refresh: bool = False,
**kwargs
) -> AnyResponse:
"""
Send HTTP request with caching.
Parameters:
- method: HTTP method
- url: Request URL
- headers: Request headers
- expire_after: Override default expiration for this request
- only_if_cached: Return 504 if not cached instead of making request
- refresh: Revalidate cached response before use
- force_refresh: Always make new request, overwrite cache
- **kwargs: Additional arguments passed to requests
"""
def send(
self,
request: PreparedRequest,
expire_after: ExpirationTime = None,
only_if_cached: bool = False,
refresh: bool = False,
force_refresh: bool = False,
**kwargs
) -> AnyResponse:
"""Send prepared request with caching."""
@property
def settings(self) -> CacheSettings:
"""Current cache settings."""
@settings.setter
def settings(self, value: CacheSettings): ...
@property
def expire_after(self) -> ExpirationTime:
"""Default expiration time (backwards compatibility)."""
@expire_after.setter
def expire_after(self, value: ExpirationTime): ...
def cache_disabled(self):
"""Context manager to temporarily disable caching."""
def close(self):
"""Close session and backend connections."""Basic usage with default SQLite backend:
from requests_cache import CachedSession
# Create session with default settings
session = CachedSession('my_cache')
response = session.get('https://api.example.com/data')
print(f"From cache: {response.from_cache}")
# Same request will be served from cache
response2 = session.get('https://api.example.com/data')
print(f"From cache: {response2.from_cache}") # TrueAdvanced configuration:
from requests_cache import CachedSession
from datetime import timedelta
session = CachedSession(
cache_name='advanced_cache',
backend='redis',
expire_after=timedelta(hours=1),
urls_expire_after={
'*.example.com/api/data': timedelta(minutes=5),
'*.slow-api.com': timedelta(days=1),
},
allowable_codes=[200, 201, 404],
allowable_methods=['GET', 'POST'],
ignored_parameters=['api_key', 'session_id'],
cache_control=True,
stale_if_error=True
)Per-request cache control:
# Override expiration for specific request
response = session.get(
'https://api.example.com/data',
expire_after=300 # 5 minutes
)
# Force refresh (bypass cache)
response = session.get(
'https://api.example.com/data',
force_refresh=True
)
# Only return if cached
response = session.get(
'https://api.example.com/data',
only_if_cached=True
)Mixin class that can add caching features to any existing requests.Session subclass or instance. Useful for integrating with custom session classes or third-party libraries.
class CacheMixin:
def __init__(self, **kwargs):
"""Initialize mixin with cache settings."""
@classmethod
def wrap(cls, original_session: Session, **kwargs) -> 'CacheMixin':
"""
Add caching to existing Session while preserving settings.
Parameters:
- original_session: Session object to wrap
- **kwargs: Cache configuration options
Returns:
CacheMixin instance with original session settings
"""
def cache_disabled(self):
"""Context manager to temporarily disable caching."""
def close(self):
"""Close session and backend connections."""Wrapping an existing session:
import requests
from requests_cache import CacheMixin
# Create and configure a regular session
original = requests.Session()
original.auth = ('username', 'password')
original.headers.update({'User-Agent': 'MyApp/1.0'})
# Add caching while preserving all settings
cached_session = CacheMixin.wrap(
original,
cache_name='wrapped_cache',
expire_after=3600
)
response = cached_session.get('https://api.example.com/data')Using as a mixin with custom session class:
from requests import Session
from requests_cache import CacheMixin
class CustomCachedSession(CacheMixin, Session):
def __init__(self, **cache_kwargs):
super().__init__(**cache_kwargs)
# Add custom session configuration
self.headers.update({'Accept': 'application/json'})
session = CustomCachedSession(
cache_name='custom_cache',
backend='filesystem'
)Additional session methods for managing cached data:
# Through session.cache property
session.cache.clear() # Clear all cached responses
session.cache.delete(urls=['https://example.com']) # Delete specific URLs
session.cache.delete_expired() # Delete expired responses only
len(session.cache) # Get number of cached responses# Session-related types
AnyResponse = Union[OriginalResponse, CachedResponse]
MutableMapping = Dict[str, str] # For headers parameterInstall 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