"""Utilities for patching ``requests``.
.. warning:: These functions are not thread-safe. Use :py:class:`.CachedSession` directly if you
want to use caching in a multi-threaded environment.
"""
from contextlib import contextmanager
from logging import getLogger
from typing import Callable, Dict, Iterable, Optional, Type
import requests
from .backends import BackendSpecifier, BaseCache
from .response import ExpirationTime
from .session import CachedSession, OriginalSession
logger = getLogger(__name__)
[docs]def install_cache(
cache_name: str = 'http_cache',
backend: BackendSpecifier = None,
expire_after: ExpirationTime = -1,
urls_expire_after: Dict[str, ExpirationTime] = None,
allowable_codes: Iterable[int] = (200,),
allowable_methods: Iterable['str'] = ('GET', 'HEAD'),
filter_fn: Callable = None,
old_data_on_error: bool = False,
session_factory: Type[OriginalSession] = CachedSession,
**kwargs,
):
"""
Install the cache for all ``requests`` functions by monkey-patching :py:class:`requests.Session`
Example:
>>> requests_cache.install_cache('demo_cache')
Accepts all the same parameters as :py:class:`.CachedSession`. Additional parameters:
Args:
session_factory: Session class to use. It must inherit from either
:py:class:`.CachedSession` or :py:class:`.CacheMixin`
"""
class _ConfiguredCachedSession(session_factory):
def __init__(self):
super().__init__(
cache_name=cache_name,
backend=backend,
expire_after=expire_after,
urls_expire_after=urls_expire_after,
allowable_codes=allowable_codes,
allowable_methods=allowable_methods,
filter_fn=filter_fn,
old_data_on_error=old_data_on_error,
**kwargs,
)
_patch_session_factory(_ConfiguredCachedSession)
[docs]def uninstall_cache():
"""Disable the cache by restoring the original :py:class:`requests.Session`"""
_patch_session_factory(OriginalSession)
[docs]@contextmanager
def disabled():
"""
Context manager for temporarily disabling caching for all ``requests`` functions
Example:
>>> with requests_cache.disabled():
... requests.get('http://httpbin.org/get')
"""
previous = requests.Session
uninstall_cache()
try:
yield
finally:
_patch_session_factory(previous)
[docs]@contextmanager
def enabled(*args, **kwargs):
"""
Context manager for temporarily enabling caching for all ``requests`` functions
Accepts the same arguments as :py:func:`.install_cache`.
Example:
>>> with requests_cache.enabled('cache_db'):
... requests.get('http://httpbin.org/get')
"""
install_cache(*args, **kwargs)
try:
yield
finally:
uninstall_cache()
[docs]def get_cache() -> Optional[BaseCache]:
"""Get the internal cache object from the currently installed ``CachedSession`` (if any)"""
return requests.Session().cache if is_installed() else None
[docs]def is_installed() -> bool:
"""Indicate whether or not requests-cache is currently installed"""
return isinstance(requests.Session(), CachedSession)
[docs]def clear():
"""Clear the currently installed cache (if any)"""
if get_cache():
get_cache().clear()
[docs]def remove_expired_responses(expire_after: ExpirationTime = None):
"""Remove expired responses from the cache, optionally with revalidation
Args:
expire_after: A new expiration time used to revalidate the cache
"""
if is_installed():
return requests.Session().remove_expired_responses(expire_after)
def _patch_session_factory(session_factory: Type[OriginalSession] = CachedSession):
logger.debug(f'Patching requests.Session with class: {session_factory.__name__}')
requests.Session = requests.sessions.Session = session_factory # noqa