How to use the cachecontrol.adapter.CacheControlAdapter function in CacheControl

To help you get started, we’ve selected a few CacheControl examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github ionrock / cachecontrol / tests / test_max_age.py View on Github external
def sess(self, url):
        self.url = url
        self.cache = DictCache()
        sess = Session()
        sess.mount(
            "http://", CacheControlAdapter(self.cache, serializer=NullSerializer())
        )
        return sess
github ionrock / cachecontrol / tests / test_adapter.py View on Github external
def use_adapter():
    print("Using adapter")
    sess = Session()
    sess.mount("http://", CacheControlAdapter())
    return sess
github ionrock / cachecontrol / tests / test_adapter.py View on Github external
def test_close(self):
        cache = mock.Mock(spec=DictCache)
        sess = Session()
        sess.mount("http://", CacheControlAdapter(cache))

        sess.close()
        assert cache.close.called
github ionrock / cachecontrol / cachecontrol / _cmd.py View on Github external
def get_session():
    adapter = CacheControlAdapter(
        DictCache(), cache_etags=True, serializer=None, heuristic=None
    )
    sess = requests.Session()
    sess.mount("http://", adapter)
    sess.mount("https://", adapter)

    sess.cache_controller = adapter.controller
    return sess
github freeipa / freeipa-pr-ci / github / open_close_pr.py View on Github external
def __init__(self, github_token, repo, args):
        github = github3.login(token=github_token)
        github.session.mount('https://api.github.com', CacheControlAdapter())
        self.repo = github.repository(repo['owner'], repo['name'])
        self.upstream_repo = github.repository('freeipa', 'freeipa')
        self.args = args
github ionrock / cachecontrol / cachecontrol / adapter.py View on Github external
self.controller.cache_response, request, response
                    ),
                )
                if response.chunked:
                    super_update_chunk_length = response._update_chunk_length

                    def _update_chunk_length(self):
                        super_update_chunk_length()
                        if self.chunk_left == 0:
                            self._fp._close()

                    response._update_chunk_length = types.MethodType(
                        _update_chunk_length, response
                    )

        resp = super(CacheControlAdapter, self).build_response(request, response)

        # See if we should invalidate the cache.
        if request.method in self.invalidating_methods and resp.ok:
            cache_url = self.controller.cache_url(request.url)
            self.cache.delete(cache_url)

        # Give the request a from_cache attr to let people use it
        resp.from_cache = from_cache

        return resp
github freeipa / freeipa-pr-ci / github / prci_github / adapter.py View on Github external
import datetime
import logging
import requests
import time

from cachecontrol.adapter import CacheControlAdapter


RETRY_TIME = 1

logger = logging.getLogger(__name__)  # pylint: disable=invalid-name


class GitHubAdapter(CacheControlAdapter):
    """Handles GitHub request rate limit exhaustion and uses cachecontrol
    """
    def __init__(self, tries=3, *args, **kwargs):
        if not isinstance(tries, int) or tries <= 0:
            raise ValueError('tries must be positive integer')
        self.tries = tries

        super(GitHubAdapter, self).__init__(*args, **kwargs)

    def send(self, request, *args, **kwargs):
        # Force the caching mechanism to ignore max-age and send the request
        # with ETag. This way we get the freshest data without consuming
        # rate-limited requests
        # TODO: once cachecontrol containing fix from
        # https://github.com/ionrock/cachecontrol/pull/163
        # is available everywhere we care replace with following
github pudo-attic / scrapekit / scrapekit / http.py View on Github external
def make_session(scraper):
    """ Instantiate a session with the desired configuration parameters,
    including the cache policy. """
    cache_path = os.path.join(scraper.config.data_path, 'cache')
    cache_policy = scraper.config.cache_policy
    cache_policy = cache_policy.lower().strip()
    session = ScraperSession()
    session.scraper = scraper
    session.cache_policy = cache_policy

    adapter = CacheControlAdapter(
        FileCache(cache_path),
        cache_etags=True,
        controller_class=PolicyCacheController
    )
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session