Skip to content

Instantly share code, notes, and snippets.

@hirokiky
Last active February 1, 2016 08:05
Show Gist options
  • Save hirokiky/4a4eaff3fdbab8561e44 to your computer and use it in GitHub Desktop.
Save hirokiky/4a4eaff3fdbab8561e44 to your computer and use it in GitHub Desktop.
requests.Session to memoize response forcedly, depeding on cachetools library.
import cachetools
import requests
class ForceCacheSession(requests.Session):
""" Force caching requests.Session
* Apply `cachetools.Cache` by `mount_cache()` methoud
* It will cache if the method was GET and status code was 200
* It will apply `res.from_cache` bool value
"""
def __init__(self):
super(ForceCacheSession, self).__init__()
self.cache = None
def mount_cache(self, cache):
self.cache = cache
def request(self, method, *args, **kwargs):
k = cachetools.typedkey(*args, **kwargs)
try:
res = self.cache[k]
res.from_cache = True
return res
except KeyError:
pass
res = super(ForceCacheSession, self).request(method, *args, **kwargs)
if method == 'GET' and res.status_code == 200:
try:
self.cache[k] = res
except ValueError:
pass # Value too large
res.from_cache = False
return res
import unittest
import responses
class TestForceCacheSession(unittest.TestCase):
def _makeOne(self):
import cachetools
from . import ForceCacheSession
target = ForceCacheSession()
target.mount_cache(cachetools.Cache(maxsize=1))
return target
@responses.activate
def test__it(self):
responses.add(responses.GET, 'http://example.com/',
status=200, body="Hello")
target = self._makeOne()
actual = target.get('http://example.com/')
self.assertEqual(actual.text, 'Hello')
self.assertFalse(actual.from_cache)
actual = target.get('http://example.com/')
self.assertEqual(actual.text, 'Hello')
self.assertTrue(actual.from_cache)
@responses.activate
def test__not_get(self):
responses.add(responses.POST, 'http://example.com/',
status=200, body="Hello")
target = self._makeOne()
actual = target.post('http://example.com/')
self.assertEqual(actual.text, 'Hello')
self.assertFalse(actual.from_cache)
actual = target.post('http://example.com/')
self.assertEqual(actual.text, 'Hello')
self.assertFalse(actual.from_cache)
@responses.activate
def test__not_200(self):
responses.add(responses.GET, 'http://example.com/',
status=400, body="Hello")
target = self._makeOne()
actual = target.get('http://example.com/')
self.assertEqual(actual.text, 'Hello')
self.assertFalse(actual.from_cache)
actual = target.get('http://example.com/')
self.assertEqual(actual.text, 'Hello')
self.assertFalse(actual.from_cache)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment