Compatibility with other libraries#

This library works by patching and/or extending requests.Session. Many other libraries out there do the same thing, making it potentially difficult to combine them.

For that scenario, a mixin class is provided, so you can create a custom class with behavior from multiple Session-modifying libraries:

>>> from requests import Session
>>> from requests_cache import CacheMixin
>>> from some_other_lib import SomeOtherMixin

>>> class CustomSession(CacheMixin, SomeOtherMixin, Session):
...     """Session class with features from both some_other_lib and requests-cache"""

Requests-HTML#

requests-html is one library that works with this method:

>>> import requests
>>> from requests_cache import CacheMixin, install_cache
>>> from requests_html import HTMLSession

>>> class CachedHTMLSession(CacheMixin, HTMLSession):
...     """Session with features from both CachedSession and HTMLSession"""

>>> session = CachedHTMLSession()
>>> response = session.get('https://github.com/')
>>> print(response.from_cache, response.html.links)

Or if you are using install_cache(), you can use the session_factory argument:

>>> install_cache(session_factory=CachedHTMLSession)
>>> response = requests.get('https://github.com/')
>>> print(response.from_cache, response.html.links)

The same approach can be used with other libraries that subclass requests.Session.

Requests-Futures#

Some libraries, including requests-futures, support wrapping an existing session object:

>>> from requests_cache import CachedSession
>>> from requests_futures.sessions import FuturesSession

>>> session = FuturesSession(session=CachedSession())

In this case, FuturesSession must wrap CachedSession rather than the other way around, since FuturesSession returns (as you might expect) futures rather than response objects. See issue #135 for more notes on this.

Requests-OAuthlib#

Usage with requests-oauthlib is the same as other libraries that subclass requests.Session:

>>> from requests_cache import CacheMixin
>>> from requests_oauthlib import OAuth2Session

>>> class CachedOAuth2Session(CacheMixin, OAuth2Session):
...     """Session with features from both CachedSession and OAuth2Session"""

>>> session = CachedOAuth2Session('my_client_id')

Requests-Ratelimiter#

requests-ratelimiter adds rate-limiting to requests via the pyrate-limiter library. It also provides a mixin, but note that the inheritance order is important: If rate-limiting is applied after caching, you get the added benefit of not counting cache hits against your rate limit.

>>> from pyrate_limiter import RedisBucket, RequestRate, Duration
>>> from requests import Session
>>> from requests_cache import CacheMixin, RedisCache
>>> from requests_ratelimiter import LimiterMixin

>>> class CachedLimiterSession(CacheMixin, LimiterMixin, Session):
...     """Session class with caching and rate-limiting behavior. Accepts arguments for both
...     LimiterSession and CachedSession.
...     """

>>> # Limit non-cached requests to 5 requests per second, with unlimited cached requests
>>> # Optionally use Redis as both the bucket backend and the cache backend
>>> session = CachedLimiterSession(
...     per_second=5,
...     bucket_class=RedisBucket,
...     backend=RedisCache(),
... )

Internet Archive#

Usage with internetarchive is the same as other libraries that subclass requests.Session:

>>> from requests_cache import CacheMixin
>>> from internetarchive.session import ArchiveSession

>>> class CachedArchiveSession(CacheMixin, ArchiveSession):
...     """Session with features from both CachedSession and ArchiveSession"""

>>> session = CachedArchiveSession()

Requests-Mock#

requests-mock has multiple methods for mocking requests, including a contextmanager, decorator, fixture, and adapter. There are a few different options for using it with requests-cache, depending on how you want your tests to work.

Disabling requests-cache#

If you have an application that uses requests-cache and you just want to use requests-mock in your tests, the easiest thing to do is to disable requests-cache.

For example, if you are using install_cache() in your application and the requests-mock pytest fixture in your tests, you could wrap it in another fixture that uses uninstall_cache() or disabled():

Example
"""Example of using requests-cache with the requests-mock library"""
import pytest
import requests

import requests_cache


@pytest.fixture(scope='function')
def requests_cache_mock(requests_mock):
    with requests_cache.disabled():
        yield requests_mock


def test_requests_cache_mock(requests_cache_mock):
    """Within this test function, requests will be mocked and not cached"""
    url = 'https://example.com'
    requests_cache_mock.get(url, text='Mock response!')

    # Make sure the mocker is used
    response_1 = requests.get(url)
    assert response_1.text == 'Mock response!'

    # Make sure the cache is not used
    response_2 = requests.get(url)
    assert getattr(response_2, 'from_cache', False) is False

Or if you use a CachedSession object, you could replace it with a regular Session, for example:

>>> import unittest
>>> import pytest
>>> import requests

>>> @pytest.fixure(scope='function', autouse=True)
>>> def disable_requests_cache():
...     """Replace CachedSession with a regular Session for all test functions"""
...     with unittest.mock.patch('requests_cache.CachedSession', requests.Session):
...         yield

Combining requests-cache with requests-mock#

If you want both caching and mocking features at the same time, you can attach requests-mock’s adapter to a CachedSession:

Example

test_requests_mock_combine_cache.py

"""Example of using requests-cache with the requests-mock library"""
import pytest
from requests_mock import Adapter

from requests_cache import CachedSession

URL = 'https://some_test_url'


@pytest.fixture(scope='function')
def mock_session():
    """Fixture that provides a CachedSession that will make mock requests where it would normally
    make real requests"""
    adapter = Adapter()
    adapter.register_uri(
        'GET',
        URL,
        headers={'Content-Type': 'text/plain'},
        text='Mock response!',
        status_code=200,
    )

    session = CachedSession(backend='memory')
    session.mount('https://', adapter)
    yield session


def test_mock_session(mock_session):
    """Test that the mock_session fixture is working as expected"""
    response_1 = mock_session.get(URL)
    assert response_1.text == 'Mock response!'
    assert getattr(response_1, 'from_cache', False) is False

    response_2 = mock_session.get(URL)
    assert response_2.text == 'Mock response!'
    assert response_2.from_cache is True

Building a mocker using requests-cache data#

Another approach is to use cached data to dynamically define mock requests + responses. This has the advantage of only using request-mock’s behavior for request matching.

@pytest.fixture(scope='session')
def mock_session():
    """Fixture that provides a session with mocked URLs and responses based on cache data"""
    adapter = Adapter()
    cache = CachedSession(TEST_DB).cache

    for response in cache.responses.values():
        adapter.register_uri(
            response.request.method,
            response.request.url,
            content=response.content,
            headers=response.headers,
            status_code=response.status_code,
        )
        print(f'Added mock response: {response}')

    session = Session()
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    yield session

To turn that into a complete example:

Example

test_requests_mock_load_cache.py

"""Example of using requests-cache with the requests-mock library"""
from os.path import dirname, join
from unittest.mock import patch

import pytest
import requests
from requests import Session
from requests_mock import Adapter, NoMockAddress

from requests_cache import CachedSession

TEST_DB = join(dirname(__file__), 'httpbin_sample.test-db')
TEST_URLS = [
    'https://httpbin.org/get',
    'https://httpbin.org/html',
    'https://httpbin.org/json',
]
UNMOCKED_URL = 'https://httpbin.org/ip'


@pytest.fixture(scope='session')
def mock_session():
    """Fixture that provides a session with mocked URLs and responses based on cache data"""
    adapter = Adapter()
    cache = CachedSession(TEST_DB).cache

    for response in cache.responses.values():
        adapter.register_uri(
            response.request.method,
            response.request.url,
            content=response.content,
            headers=response.headers,
            status_code=response.status_code,
        )
        print(f'Added mock response: {response}')

    session = Session()
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    yield session


@patch.object(requests.adapters.HTTPAdapter, 'send', side_effect=ValueError('Real request made!'))
def test_mock_session(mock_http_adapter, mock_session):
    """Test that the mock_session fixture is working as expected"""
    # An error will be raised if a real request is made
    with pytest.raises(ValueError):
        requests.get(TEST_URLS[0])

    # All mocked URLs will return a response based on requests-cache data
    for url in TEST_URLS:
        response = mock_session.get(url)
        assert getattr(response, 'from_cache', False) is False

    # requests-mock will raise an error for an unmocked URL, as usual
    with pytest.raises(NoMockAddress):
        mock_session.get(UNMOCKED_URL)


def save_test_data():
    """Run once to save data to reuse for tests, for demo purposes.
    In practice, you could just run your application or tests with requests-cache installed.
    """
    session = CachedSession(TEST_DB)
    for url in TEST_URLS:
        session.get(url)


if __name__ == '__main__':
    save_test_data()

Responses#

Usage with the responses library is similar to the requests-mock examples above.

Example

test_responses_load_cache.py

"""Example of using requests-cache with the responses library"""
from contextlib import contextmanager
from os.path import dirname, join
from unittest.mock import patch

import pytest
import requests
from requests.exceptions import ConnectionError
from responses import RequestsMock, Response

from requests_cache import CachedSession

TEST_DB = join(dirname(__file__), 'httpbin_sample.test-db')
TEST_URLS = [
    'https://httpbin.org/get',
    'https://httpbin.org/html',
    'https://httpbin.org/json',
]
PASSTHRU_URL = 'https://httpbin.org/gzip'
UNMOCKED_URL = 'https://httpbin.org/ip'


@contextmanager
def get_responses():
    """Contextmanager that provides a RequestsMock object mocked URLs and responses
    based on cache data
    """
    with RequestsMock() as mocker:
        cache = CachedSession(TEST_DB).cache
        for response in cache.responses.values():
            mocker.add(
                Response(
                    response.request.method,
                    response.request.url,
                    body=response.content,
                    headers=response.headers,
                    status=response.status_code,
                )
            )
        mocker.add_passthru(PASSTHRU_URL)
        yield mocker


# responses patches HTTPAdapter.send(), so we need to patch one level lower to verify request mocking
@patch.object(
    requests.adapters.HTTPAdapter, 'get_connection', side_effect=ValueError('Real request made!')
)
def test_mock_session(mock_http_adapter):
    """Test that the mock_session fixture is working as expected"""
    with get_responses():
        # An error will be raised if a real request is made
        with pytest.raises(ValueError):
            requests.get(PASSTHRU_URL)

        # All mocked URLs will return a response based on requests-cache data
        for url in TEST_URLS:
            response = requests.get(url)
            assert getattr(response, 'from_cache', False) is False

        # responses will raise an error for an unmocked URL, as usual
        with pytest.raises(ConnectionError):
            requests.get(UNMOCKED_URL)

VCR#

If you would like to reuse your cached response data for unit tests, one option is to convert your cache into a format compatible with VCR-vased libraries like vcrpy and betamax.

Example

vcr.py

from os import makedirs
from os.path import abspath, dirname, expanduser, join
from typing import Any, Dict, Iterable
from urllib.parse import urlparse

import yaml

from requests_cache import BaseCache, CachedResponse, CachedSession, __version__
from requests_cache.serializers.preconf import yaml_preconf_stage


def to_vcr_cassette(cache: BaseCache, path: str):
    """Export cached responses to a VCR-compatible YAML file (cassette)

    Args:
        cache: Cache instance containing response data to export
        path: Path for new cassette file
    """

    responses = cache.responses.values()
    write_cassette(to_vcr_cassette_dict(responses), path)


def to_vcr_cassettes_by_host(cache: BaseCache, cassette_dir: str = '.'):
    """Export cached responses as VCR-compatible YAML files (cassettes), split into separate files
    based on request host

    Args:
        cache: Cache instance containing response data to export
        cassette_dir: Base directory for cassette library
    """
    responses = cache.responses.values()
    for host, cassette in to_vcr_cassette_dicts_by_host(responses).items():
        write_cassette(cassette, join(cassette_dir, f'{host}.yml'))


def to_vcr_cassette_dict(responses: Iterable[CachedResponse]) -> Dict:
    """Convert responses to a VCR cassette dict"""
    return {
        'http_interactions': [to_vcr_episode(r) for r in responses],
        'recorded_with': f'requests-cache {__version__}',
    }


def to_vcr_episode(response: CachedResponse) -> Dict:
    """Convert a single response to a VCR-compatible response ("episode") dict"""
    # Do most of the work with cattrs + default YAML conversions
    response_dict = yaml_preconf_stage.dumps(response)

    def _to_multidict(d):
        return {k: [v] for k, v in d.items()}

    # Translate requests.Response structure into VCR format
    return {
        'request': {
            'body': response_dict['request']['body'],
            'headers': _to_multidict(response_dict['request']['headers']),
            'method': response_dict['request']['method'],
            'uri': response_dict['request']['url'],
        },
        'response': {
            'body': {'string': response_dict['_content'], 'encoding': response_dict['encoding']},
            'headers': _to_multidict(response_dict['headers']),
            'status': {'code': response_dict['status_code'], 'message': response_dict['reason']},
            'url': response_dict['url'],
        },
        'recorded_at': response_dict['created_at'],
    }


def to_vcr_cassette_dicts_by_host(responses: Iterable[CachedResponse]) -> Dict[str, Dict]:
    responses_by_host: Dict[str, Any] = {}
    for response in responses:
        host = urlparse(response.request.url).netloc
        responses_by_host.setdefault(host, [])
        responses_by_host[host].append(response)
    return {host: to_vcr_cassette_dict(responses) for host, responses in responses_by_host.items()}


def write_cassette(cassette, path):
    path = abspath(expanduser(path))
    makedirs(dirname(path), exist_ok=True)
    with open(path, 'w') as f:
        f.write(yaml.safe_dump(cassette))


# Create an example cache and export it to a cassette
if __name__ == '__main__':
    cache_dir = 'example_cache'
    session = CachedSession(join(cache_dir, 'http_cache.sqlite'))
    session.get('https://httpbin.org/get')
    session.get('https://httpbin.org/json')
    to_vcr_cassette(session.cache, join(cache_dir, 'http_cache.yaml'))