Consolidate expiration, pre-serializiation, and other response object logic into CachedResponse class:

* Replace `_RawStore`  with `CachedHTTPResponse` class to wrap raw responses
    * Maintain support for streaming requests (#68)
    * Improve handling for generator usage
    * Add support for use with `pandas.read_csv()` and similar readers (#148)
    * Add support for use as a context manager (#148)
    * Add support for `decode_content` arg
* Fix streaming requests when used with memory backend (#188)
* Verified that `PreparedRequest.body` is always encoded in utf-8, so no need to detect encoding (Re: TODO note)
* Response creation time and expiration time are stored as CachedResponse, so the `(response, timestamp)` tuple is no longer necessary
* Rename `response.expire_after` and `response.cache_date` to `expires` and `created_at`, respectively, based on browser cache directives
* Add optional `expire_after` param to `CachedSession.remove_old_responses()`
* Make `CachedSession` members `allowable_codes, allowable_methods, filter_fn, old_data_on_error`
  public, since they can safely be modified after initialization
* More type annotations and docstring updates
* Move main cache documentation from `CacheMixin` to CachedSession`, since that's probably where a user would look first
* Wrap temporary `_request_expire_after` in a contextmanager
* Add intersphinx links for `urllib` classes & methods
* Fix linting issues raised by flake8
* Start adding some unit tests using requests-mock

tmp

5 files changed, 452 insertions, 349 deletions

diff --git a/requests_cache/__init__.py b/requests_cache/__init__.py
index 067506f..c4b5878 100644
--- a/requests_cache/__init__.py
+++ b/requests_cache/__init__.py
@@ -1,4 +1,5 @@
 #!/usr/bin/env python
+# flake8: noqa: E402,F401
 """
     requests_cache
     ~~~~~~~~~~~~~~
@@ -18,15 +19,17 @@
         # will take approximately 5 seconds instead 50
 
 
-    :copyright: (c) 2012 by Roman Haritonov.
+    :copyright: (c) 2021 by Roman Haritonov.
     :license: BSD, see LICENSE for more details.
 """
 __docformat__ = 'restructuredtext'
 __version__ = '0.6.0'
 
-# Quietly ignore importerror, if setup.py is invoked outside a virtualenv
+# Quietly ignore ImportError, if setup.py is invoked outside a virtualenv
 try:
+    from .response import AnyResponse, CachedHTTPResponse, CachedResponse, ExpirationTime
     from .core import (
+        ALL_METHODS,
         CachedSession,
         CacheMixin,
         clear,
diff --git a/requests_cache/backends/__init__.py b/requests_cache/backends/__init__.py
index 178ef08..6d1282d 100644
--- a/requests_cache/backends/__init__.py
+++ b/requests_cache/backends/__init__.py
@@ -1,13 +1,28 @@
-# noqa: F401
+# flake8: noqa: F401
 """
     requests_cache.backends
     ~~~~~~~~~~~~~~~~~~~~~~~
 
     Classes and functions for cache persistence
 """
-
-
-from .base import BACKEND_KWARGS, BaseCache
+from .base import BaseCache
+
+# All backend-specific keyword arguments combined
+BACKEND_KWARGS = [
+    'connection',
+    'db_name',
+    'endpont_url',
+    'extension',
+    'fast_save',
+    'ignored_parameters',
+    'include_get_headers',
+    'location',
+    'name',
+    'namespace',
+    'read_capacity_units',
+    'region_name',
+    'write_capacity_units',
+]
 
 registry = {
     'memory': BaseCache,
diff --git a/requests_cache/backends/base.py b/requests_cache/backends/base.py
index f97d522..5698c4a 100644
--- a/requests_cache/backends/base.py
+++ b/requests_cache/backends/base.py
@@ -1,4 +1,3 @@
-#!/usr/bin/env python
 """
     requests_cache.backends.base
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -7,34 +6,19 @@
     extended to support persistence.
 """
 import hashlib
-from copy import copy
-from datetime import datetime, timezone
-from io import BytesIO
+import json
+from pickle import PickleError
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 
 import requests
 
-# All backend-specific keyword arguments combined
-BACKEND_KWARGS = [
-    'connection',
-    'db_name',
-    'endpont_url',
-    'extension',
-    'fast_save',
-    'ignored_parameters',
-    'include_get_headers',
-    'location',
-    'name',
-    'namespace',
-    'read_capacity_units',
-    'region_name',
-    'write_capacity_units',
-]
+from ..response import AnyResponse, CachedResponse, ExpirationTime
+
 DEFAULT_HEADERS = requests.utils.default_headers()
 
 
 class BaseCache(object):
-    """Base class for cache implementations, can be used as in-memory cache.
+    """Base class for cache implementations, which can also be used as in-memory cache.
 
     To extend it you can provide dictionary-like objects for
     :attr:`keys_map` and :attr:`responses` or override public methods.
@@ -48,239 +32,151 @@ class BaseCache(object):
         self._include_get_headers = kwargs.get("include_get_headers", False)
         self._ignored_parameters = set(kwargs.get("ignored_parameters") or [])
 
-    def save_response(self, key, response):
+    def save_response(self, key: str, response: AnyResponse, expire_after: ExpirationTime = None):
         """Save response to cache
 
-        :param key: key for this response
-        :param response: response to save
-
-        .. note:: Response is reduced before saving (with :meth:`reduce_response`)
-                  to make it picklable
+        Args:
+            key: key for this response
+            response: response to save
+            expire_after: Time in seconds until this cache item should expire
         """
-        self.responses[key] = self.reduce_response(response), datetime.now(timezone.utc)
+        self.responses[key] = CachedResponse(response, expire_after=expire_after)
 
-    def add_key_mapping(self, new_key, key_to_response):
+    def add_key_mapping(self, new_key: str, key_to_response: str):
         """
         Adds mapping of `new_key` to `key_to_response` to make it possible to
         associate many keys with single response
 
-        :param new_key: new key (e.g. url from redirect)
-        :param key_to_response: key which can be found in :attr:`responses`
-        :return:
+        Args:
+            new_key: New resource key (e.g. url from redirect)
+            key_to_response: Key which can be found in :attr:`responses`
         """
         self.keys_map[new_key] = key_to_response
 
-    def get_response_and_time(self, key, default=(None, None)):
-        """Retrieves response and timestamp for `key` if it's stored in cache,
-        otherwise returns `default`
+    def get_response(self, key: str, default=None) -> CachedResponse:
+        """Retrieves response for `key` if it's stored in cache, otherwise returns `default`
 
-        :param key: key of resource
-        :param default: return this if `key` not found in cache
-        :returns: tuple (response, datetime)
-
-        .. note:: Response is restored after unpickling with :meth:`restore_response`
+        Args:
+            key: Key of resource
+            default: Value to return if `key` is not in cache
         """
         try:
             if key not in self.responses:
                 key = self.keys_map[key]
-            response, timestamp = self.responses[key]
-        except KeyError:
+            response = self.responses[key]
+            response.reset()  # In case response was in memory and raw content has already been read
+            return response
+        except (KeyError, TypeError, PickleError):
             return default
-        return self.restore_response(response), timestamp
 
-    def delete(self, key):
+    def delete(self, key: str):
         """Delete `key` from cache. Also deletes all responses from response history"""
         try:
             if key in self.responses:
-                response, _ = self.responses[key]
+                response = self.responses[key]
                 del self.responses[key]
             else:
-                response, _ = self.responses[self.keys_map[key]]
+                response = self.responses[self.keys_map[key]]
                 del self.keys_map[key]
             for r in response.history:
                 del self.keys_map[self.create_key(r.request)]
         except KeyError:
             pass
 
-    def delete_url(self, url):
+    def delete_url(self, url: str):
         """Delete response associated with `url` from cache.
         Also deletes all responses from response history. Works only for GET requests
         """
         self.delete(self._url_to_key(url))
 
     def clear(self):
-        """Clear cache"""
+        """Delete all items from the cache"""
         self.responses.clear()
         self.keys_map.clear()
 
-    def remove_old_entries(self):
-        """Deletes expired entries from the cache"""
-        keys_to_delete = set()
-        for key, (response, _) in self.responses.items():
-            if _is_expired(response):
-                keys_to_delete.add(key)
+    def remove_expired_responses(self, expire_after: ExpirationTime = None):
+        """Remove expired responses from the cache, optionally with revalidation
 
-        for key in keys_to_delete:
-            self.delete(key)
-
-    def has_key(self, key):
+        Args:
+            expire_after: A new expiration time used to revalidate the cache
+        """
+        for key, response in list(self.responses.items()):
+            # If we're revalidating and it's not yet expired, update the cached item's expiration
+            if expire_after is not None and not response.revalidate(expire_after):
+                self.responses[key] = response
+            if response.is_expired:
+                self.delete(key)
+
+    def has_key(self, key: str) -> bool:
         """Returns `True` if cache has `key`, `False` otherwise"""
         return key in self.responses or key in self.keys_map
 
-    def has_url(self, url):
-        """Returns `True` if cache has `url`, `False` otherwise.
-        Works only for GET request urls
-        """
-        return self.has_key(self._url_to_key(url))
+    def has_url(self, url: str) -> bool:
+        """Returns `True` if cache has `url`, `False` otherwise. Works only for GET request urls"""
+        return self.has_key(self._url_to_key(url))  # noqa: W601
 
-    def _url_to_key(self, url):
+    def _url_to_key(self, url: str) -> str:
         session = requests.Session()
         return self.create_key(session.prepare_request(requests.Request('GET', url)))
 
-    _response_attrs = [
-        '_content',
-        'url',
-        'status_code',
-        'cookies',
-        'headers',
-        'encoding',
-        'request',
-        'reason',
-        'raw',
-        'expiration_date',
-        'expire_after',
-    ]
-
-    _raw_response_attrs = [
-        '_original_response',
-        'decode_content',
-        'headers',
-        'reason',
-        'status',
-        'strict',
-        'version',
-    ]
-
-    def reduce_response(self, response, seen=None):
-        """Reduce response object to make it compatible with ``pickle``"""
-        if seen is None:
-            seen = {}
-        try:
-            return seen[id(response)]
-        except KeyError:
-            pass
-        result = _Store()
-        # prefetch
-        content = response.content
-        for field in self._response_attrs:
-            setattr(result, field, self._picklable_field(response, field))
-        seen[id(response)] = result
-        result.history = tuple(self.reduce_response(r, seen) for r in response.history)
-        # Emulate stream fp is not consumed yet. See #68
-        if response.raw is not None:
-            response.raw._fp = BytesIO(content)
-        return result
-
-    def _picklable_field(self, response, name):
-        value = getattr(response, name, None)
-        if name == 'request':
-            value = copy(value)
-            value.hooks = []
-        elif name == 'raw':
-            result = _RawStore()
-            for field in self._raw_response_attrs:
-                setattr(result, field, getattr(value, field, None))
-            if result._original_response is not None:
-                setattr(result._original_response, "fp", None)  # _io.BufferedReader is not picklable
-            value = result
-        return value
-
-    def restore_response(self, response, seen=None):
-        """Restore response object after unpickling"""
-        if seen is None:
-            seen = {}
-        try:
-            return seen[id(response)]
-        except KeyError:
-            pass
-        result = requests.Response()
-        for field in self._response_attrs:
-            setattr(result, field, getattr(response, field, None))
-        result.raw._cached_content_ = result.content
-        seen[id(response)] = result
-        result.history = tuple(self.restore_response(r, seen) for r in response.history)
-        return result
-
-    def _remove_ignored_parameters(self, request):
-        def filter_ignored_parameters(data):
-            return [(k, v) for k, v in data if k not in self._ignored_parameters]
-
-        url = urlparse(request.url)
-        query = parse_qsl(url.query)
-        query = filter_ignored_parameters(query)
-        query = urlencode(query)
-        url = urlunparse((url.scheme, url.netloc, url.path, url.params, query, url.fragment))
-        body = request.body
-        content_type = request.headers.get('content-type')
-        if body and content_type:
-            if content_type == 'application/x-www-form-urlencoded':
-                body = parse_qsl(body)
-                body = filter_ignored_parameters(body)
-                body = urlencode(body)
-            elif content_type == 'application/json':
-                import json
-
-                if isinstance(body, bytes):
-                    body = str(body, "utf8")  # TODO how to get body encoding?
-                body = json.loads(body)
-                body = filter_ignored_parameters(sorted(body.items()))
-                body = json.dumps(body)
-        return url, body
-
-    def create_key(self, request):
-        if self._ignored_parameters:
-            url, body = self._remove_ignored_parameters(request)
-        else:
-            url, body = request.url, request.body
+    def create_key(self, request: requests.PreparedRequest) -> str:
+        url = self._remove_ignored_url_parameters(request)
+        body = self._remove_ignored_body_parameters(request)
         key = hashlib.sha256()
-        key.update(_to_bytes(request.method.upper()))
-        key.update(_to_bytes(url))
-        if request.body:
-            key.update(_to_bytes(body))
+        key.update(_encode(request.method.upper()))
+        key.update(_encode(url))
+
+        if body:
+            key.update(_encode(body))
         else:
             if self._include_get_headers and request.headers != DEFAULT_HEADERS:
                 for name, value in sorted(request.headers.items()):
-                    key.update(_to_bytes(name))
-                    key.update(_to_bytes(value))
+                    key.update(_encode(name))
+                    key.update(_encode(value))
         return key.hexdigest()
 
-    def __str__(self):
-        return 'keys: %s\nresponses: %s' % (self.keys_map, self.responses)
-
-
-# used for saving response attributes
-class _Store(object):
-    pass
+    def _remove_ignored_url_parameters(self, request: requests.PreparedRequest) -> str:
+        url = str(request.url)
+        if not self._ignored_parameters:
+            return url
 
+        url = urlparse(url)
+        query = parse_qsl(url.query)
+        query = self._filter_ignored_parameters(query)
+        query = urlencode(query)
+        url = urlunparse((url.scheme, url.netloc, url.path, url.params, query, url.fragment))
+        return url
 
-class _RawStore(object):
-    # noop for cached response
-    def release_conn(self):
-        pass
+    def _remove_ignored_body_parameters(self, request: requests.PreparedRequest) -> str:
+        body = request.body
+        content_type = request.headers.get('content-type')
+        if not self._ignored_parameters or not body or not content_type:
+            return request.body
+
+        if content_type == 'application/x-www-form-urlencoded':
+            body = parse_qsl(body)
+            body = self._filter_ignored_parameters(body)
+            body = urlencode(body)
+        elif content_type == 'application/json':
+            body = json.loads(_decode(body))
+            body = self._filter_ignored_parameters(sorted(body.items()))
+            body = json.dumps(body)
+        return body
+
+    def _filter_ignored_parameters(self, data):
+        return [(k, v) for k, v in data if k not in self._ignored_parameters]
 
-    # for streaming requests support
-    def read(self, chunk_size=1):
-        if not hasattr(self, "_io_with_content_"):
-            self._io_with_content_ = BytesIO(self._cached_content_)
-        return self._io_with_content_.read(chunk_size)
+    def __str__(self):
+        return f'redirects: {len(self.keys_map)}\nresponses: {len(self.responses)}'
 
 
-def _is_expired(response):
-    """Check a cached response to see if it's expired"""
-    if getattr(response, 'expire_after', None) is not None:
-        return datetime.now(timezone.utc) > response.expire_after
-    return False
+def _encode(value, encoding='utf-8') -> bytes:
+    """Encode a value, if it hasn't already been"""
+    return value if isinstance(value, bytes) else value.encode(encoding)
 
 
-def _to_bytes(s, encoding='utf-8'):
-    return s if isinstance(s, bytes) else bytes(s, encoding)
+def _decode(value, encoding='utf-8') -> str:
+    """Decode a value, if hasn't already been.
+    Note: PreparedRequest.body is always encoded in utf-8.
+    """
+    return value if isinstance(value, str) else value.decode(encoding)
diff --git a/requests_cache/core.py b/requests_cache/core.py
index af60a4b..2a4c985 100644
--- a/requests_cache/core.py
+++ b/requests_cache/core.py
@@ -6,58 +6,30 @@
 """
 from collections.abc import Mapping
 from contextlib import contextmanager
-from datetime import datetime, timedelta, timezone
 from operator import itemgetter
-from typing import Any, Callable, Dict, Iterable, Optional, Union
+from typing import Any, Callable, Dict, Iterable, Optional, Type
 
 import requests
+from requests import PreparedRequest
 from requests import Session as OriginalSession
 from requests.hooks import dispatch_hook
 
-from requests_cache.backends.base import BACKEND_KWARGS, _is_expired
-
 from . import backends
+from .response import AnyResponse, ExpirationTime, set_response_defaults
 
-ExpirationTime = Union[None, int, float, datetime, timedelta]
+ALL_METHODS = ['GET', 'HEAD', 'OPTIONS', 'POST', 'PUT', 'PATCH', 'DELETE']
 
 
 class CacheMixin:
-    """Mixin class that extends ``requests.Session`` with caching features.
-
-    Args:
-        cache_name: Cache prefix or namespace, depending on backend; see notes below
-        backend: Cache backend name; one of ``['sqlite', 'mongodb', 'gridfs', 'redis', 'dynamodb', 'memory']``.
-                Default behavior is to use ``'sqlite'`` if available, otherwise fallback to ``'memory'``.
-        expire_after: Number of seconds after which a cache entry will expire; set to ``None`` to
-            never expire
-        allowable_codes: Only cache responses with one of these codes
-        allowable_methods: Cache only responses for one of these HTTP methods
-        include_get_headers: Make request headers part of the cache key
-        ignored_parameters: List of request parameters to be excluded from the cache key.
-        filter_fn: function that takes a :py:class:`aiohttp.ClientResponse` object and
-            returns a boolean indicating whether or not that response should be cached. Will be
-            applied to both new and previously cached responses
-        old_data_on_error: Return expired cached responses if new request fails
-
-    See individual backend classes for additional backend-specific arguments.
-
-    The ``cache_name`` parameter will be used as follows depending on the backend:
-
-        * ``sqlite``: Cache filename prefix, e.g ``my_cache.sqlite``
-        * ``mongodb``: Database name
-        * ``redis``: Namespace, meaning all keys will be prefixed with ``'cache_name:'``
-
-    Note on cache key parameters: Set ``include_get_headers=True`` if you want responses to be
-    cached under different keys if they only differ by headers. You may also provide
-    ``ignored_parameters`` to ignore specific request params. This is useful, for example, when
-    requesting the same resource with different credentials or access tokens.
+    """Mixin class that extends :py:class:`requests.Session` with caching features.
+    See :py:class:`.CachedSession` for usage information.
     """
 
     def __init__(
         self,
         cache_name: str = 'cache',
         backend: str = None,
-        expire_after: Union[int, float, timedelta] = None,
+        expire_after: ExpirationTime = None,
         allowable_codes: Iterable[int] = (200,),
         allowable_methods: Iterable['str'] = ('GET', 'HEAD'),
         filter_fn: Callable = None,
@@ -65,115 +37,144 @@ class CacheMixin:
         **kwargs,
     ):
         self.cache = backends.create_backend(backend, cache_name, kwargs)
-        self._cache_name = cache_name
-        self.expire_after = _get_timedelta(expire_after)
+        self.allowable_codes = allowable_codes
+        self.allowable_methods = allowable_methods
+        self.filter_fn = filter_fn or (lambda r: True)
+        self.old_data_on_error = old_data_on_error
 
-        self._cache_allowable_codes = allowable_codes
-        self._cache_allowable_methods = allowable_methods
-        self._filter_fn = filter_fn or (lambda r: True)
-        self._return_old_data_on_error = old_data_on_error
-        self._is_cache_disabled = False
+        self._cache_name = cache_name
+        self._expire_after = expire_after
+        self._request_expire_after: ExpirationTime = None
+        self._disabled = False
 
         # Remove any requests-cache-specific kwargs before passing along to superclass
-        session_kwargs = {k: v for k, v in kwargs.items() if k not in BACKEND_KWARGS}
+        session_kwargs = {k: v for k, v in kwargs.items() if k not in backends.BACKEND_KWARGS}
         super().__init__(**session_kwargs)
 
-    def send(self, request, **kwargs):
-        _request_expire_after = kwargs.get('params', {}).pop('_request_expire_after', None)
-        expire_after = _request_expire_after or self.expire_after
-
-        # If we shouldn't cache the response, just send the request
-        do_not_cache = self._is_cache_disabled or request.method not in self._cache_allowable_methods
-        if do_not_cache:
-            response = super().send(request, **kwargs)
-            response.from_cache = False
-            response.cache_date = None
-            response.expire_after = None
-            return response
-
-        # If a response isn't already cached, send the request and cache the response
-        cache_key = self.cache.create_key(request)
-        try:
-            response, timestamp = self.cache.get_response_and_time(cache_key)
-        except (ImportError, TypeError):
-            response, timestamp = None, None
-        if response is None:
-            return self.send_request_and_cache_response(request, cache_key, expire_after, **kwargs)
-
-        # If the cached response is invalid, send the request and cache the response
-        if _is_expired(response):
-            try:
-                new_response = self.send_request_and_cache_response(request, cache_key, expire_after, **kwargs)
-                self.cache.delete(cache_key)
-                return new_response
-            except Exception:
-                # Return the expired/invalid response on error, if specified
-                if self._return_old_data_on_error:
-                    return response
-                self.cache.delete(cache_key)
-                raise
-
-        # Dispatch hook here, because we've removed it before pickling
-        response.from_cache = True
-        response.cache_date = timestamp
-        response = dispatch_hook('response', request.hooks, response, **kwargs)
-        return response
+    @property
+    def expire_after(self):
+        """Get either the per-session expiration, or per-request expiration, if set"""
+        return self._request_expire_after or self._expire_after
 
-    def send_request_and_cache_response(self, request, cache_key, expire_after, **kwargs):
-        response = super().send(request, **kwargs)
-        response.from_cache = False
-        response.cache_date = None
+    @expire_after.setter
+    def expire_after(self, value: ExpirationTime):
+        """Set per-session expiration"""
+        self._expire_after = value
 
-        # Cache the response, if possible
-        if response.status_code in self._cache_allowable_codes:
-            response.expire_after = _get_absolute_time(expire_after)  # type: ignore
-            self.cache.save_response(cache_key, response)
-        return response
+    @contextmanager
+    def request_expire_after(self, expire_after: ExpirationTime = None):
+        """Temporarily override ``expire_after`` for an individual request"""
+        self._request_expire_after = expire_after
+        yield
+        self._request_expire_after = None
 
     def request(
         self,
         method: str,
         url: str,
-        params: dict = None,
+        params: Dict = None,
         data: Any = None,
         expire_after: ExpirationTime = None,
         **kwargs,
-    ) -> requests.Response:
+    ) -> AnyResponse:
         """This method prepares and sends a request while automatically performing any necessary
         caching operations. This will be called by any other method-specific ``requests`` functions
-        (get, post, etc.).
+        (get, post, etc.). This does not include prepared requests, which will still be cached via
+        ``send()``.
 
-        In all cases, if the value is an explicit datetime it returned as is.
-        If it is None, it is also returned as is and caches forever.
-        All other values will be considered a relative time in the future.
+        See :py:meth:`requests.Session.request` for parameters. Additional parameters:
 
         Args:
             expire_after: Expiration time to set only for this request; see details below.
-                Overrides ``CachedSession.expire_after``. Accepts all the same types as
-                ``CachedSession.expire_after`` except for ``None``; use
-                ``CachedSession.cache_disabled`` to disable caching on a per-request basis.
+                Overrides ``CachedSession.expire_after``. Accepts all the same values as
+                ``CachedSession.expire_after`` except for ``None``; use ``-1`` to disable expiration
+                on a per-request basis.
 
         Returns:
             Either a new or cached response
-        """
-        # Store expire_after to be used by send()
-        params = _normalize_parameters(params)
-        params['_request_expire_after'] = expire_after
 
-        response = super().request(method, url, params, _normalize_parameters(data), **kwargs)
-        if self._is_cache_disabled:
+        **Order of operations:** A request will pass through the following methods:
+
+        1. :py:func:`requests.get`/:py:meth:`requests.Session.get` or other method-specific functions (optional)
+        2. :py:meth:`.CachedSession.request`
+        3. :py:meth:`requests.Session.request`
+        4. :py:meth:`.CachedSession.send`
+        5. :py:meth:`.BaseCache.get_response`
+        6. :py:meth:`requests.Session.send` (if not cached)
+        """
+        with self.request_expire_after(expire_after):
+            response = super().request(
+                method,
+                url,
+                _normalize_parameters(params),
+                _normalize_parameters(data),
+                **kwargs,
+            )
+        if self._disabled:
             return response
-        main_key = self.cache.create_key(response.request)
 
-        # If self._return_old_data_on_error is set, responses may not have the from_cache attribute
-        if hasattr(response, "from_cache") and not response.from_cache and not self._filter_fn(response):
+        # If the request has been filtered out, delete previously cached response if it exists
+        main_key = self.cache.create_key(response.request)
+        if not response.from_cache and not self.filter_fn(response):
             self.cache.delete(main_key)
             return response
 
+        # Cache redirect history
         for r in response.history:
             self.cache.add_key_mapping(self.cache.create_key(r.request), main_key)
         return response
 
+    def send(self, request: PreparedRequest, **kwargs) -> AnyResponse:
+        """Send a prepared request, with caching."""
+        # If we shouldn't cache the response, just send the request
+        if not self._is_cacheable(request):
+            response = super().send(request, **kwargs)
+            return set_response_defaults(response)
+
+        # Attempt to fetch the cached response
+        cache_key = self.cache.create_key(request)
+        try:
+            response = self.cache.get_response(cache_key)
+        except (ImportError, TypeError, ValueError):
+            response = None
+
+        # Attempt to fetch and cache a new response, if needed
+        if response is None:
+            return self._send_and_cache(request, cache_key, **kwargs)
+        if response.is_expired:
+            return self._handle_expired_response(request, response, cache_key, **kwargs)
+
+        # Dispatch hook here, because we've removed it before pickling
+        return dispatch_hook('response', request.hooks, response, **kwargs)
+
+    def _is_cacheable(self, request: PreparedRequest) -> bool:
+        criteria = [
+            not self._disabled,
+            str(request.method) in self.allowable_methods,
+            self.filter_fn(request),
+        ]
+        return all(criteria)
+
+    def _handle_expired_response(self, request, response, cache_key, **kwargs) -> AnyResponse:
+        """Determine what to do with an expired response, depending on old_data_on_error setting"""
+        # Attempt to send the request and cache the new response
+        try:
+            new_response = self._send_and_cache(request, cache_key, **kwargs)
+            self.cache.delete(cache_key)
+            return new_response
+        # Return the expired/invalid response on error, if specified; otherwise reraise
+        except Exception:
+            if self.old_data_on_error:
+                return response
+            self.cache.delete(cache_key)
+            raise
+
+    def _send_and_cache(self, request, cache_key, **kwargs):
+        response = super().send(request, **kwargs)
+        if response.status_code in self.allowable_codes:
+            self.cache.save_response(cache_key, response, self.expire_after)
+        return set_response_defaults(response)
+
     @contextmanager
     def cache_disabled(self):
         """
@@ -184,52 +185,96 @@ class CacheMixin:
             >>> with s.cache_disabled():
             ...     s.get('http://httpbin.org/ip')
         """
-        if self._is_cache_disabled:
+        if self._disabled:
             yield
         else:
-            self._is_cache_disabled = True
+            self._disabled = True
             try:
                 yield
             finally:
-                self._is_cache_disabled = False
+                self._disabled = False
+
+    def remove_expired_responses(self, expire_after: ExpirationTime = None):
+        """Remove expired responses from the cache, optionally with revalidation
 
-    def remove_expired_responses(self):
-        """Removes expired responses from storage"""
-        self.cache.remove_old_entries()
+        Args:
+            expire_after: A new expiration time used to revalidate the cache
+        """
+        self.cache.remove_expired_responses(expire_after)
 
     def __repr__(self):
         return (
             f"<CachedSession({self.cache.__class__.__name__}('{self._cache_name}', ...), "
-            f"expire_after={self.expire_after}, allowable_methods={self._cache_allowable_methods})>"
+            f"expire_after={self.expire_after}, allowable_methods={self.allowable_methods})>"
         )
 
 
 class CachedSession(CacheMixin, OriginalSession):
-    pass
+    """Class that extends ``requests.Session`` with caching features.
+    See individual backend classes for additional backend-specific arguments.
+
+    Args:
+        cache_name: Cache prefix or namespace, depending on backend
+        backend: Cache backend name; one of ``['sqlite', 'mongodb', 'gridfs', 'redis', 'dynamodb', 'memory']``.
+                Default behavior is to use ``'sqlite'`` if available, otherwise fallback to ``'memory'``.
+        expire_after: Time after which cached items will expire (see notes below)
+        allowable_codes: Only cache responses with one of these codes
+        allowable_methods: Cache only responses for one of these HTTP methods
+        include_get_headers: Make request headers part of the cache key
+        ignored_parameters: List of request parameters to be excluded from the cache key
+        filter_fn: function that takes a :py:class:`aiohttp.ClientResponse` object and
+            returns a boolean indicating whether or not that response should be cached. Will be
+            applied to both new and previously cached responses.
+        old_data_on_error: Return expired cached responses if new request fails
+
+    **Cache Name:**
+
+    The ``cache_name`` parameter will be used as follows depending on the backend:
+
+    * ``sqlite``: Cache filename, e.g ``my_cache.sqlite``
+    * ``mongodb``: Database name
+    * ``redis``: Namespace, meaning all keys will be prefixed with ``'cache_name:'``
+
+    **Cache Keys:**
+
+    The cache key is a hash created from request information, and is used as an index for cached
+    responses. There are a couple ways you can customize how the cache key is created:
+
+    * Use ``include_get_headers`` if you want headers to be included in the cache key. In other
+      words, this will create separate cache items for responses with different headers.
+    * Use ``ignored_parameters`` to exclude specific request params from the cache key. This is
+      useful, for example, if you request the same resource with different credentials or access
+      tokens.
+
+    **Cache Expiration:**
+
+    Use ``expire_after`` to specify how long responses will be cached. This can be a number
+    (in seconds), a :py:class:`.timedelta`, or a :py:class:`datetime`. Use ``None`` or ``-1`` to
+    never expire. This will not apply to responses cached in the current session; to apply a
+    different expiration to previously cached responses, see :py:meth:`remove_expired_responses`.
+    """
 
 
 def install_cache(
     cache_name: str = 'cache',
     backend: str = None,
-    expire_after: Union[int, float, timedelta] = None,
+    expire_after: 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=CachedSession,
+    session_factory: Type[OriginalSession] = CachedSession,
     **kwargs,
 ):
     """
-    Installs cache for all ``Requests`` requests by monkey-patching ``Session``
+    Installs cache for all ``requests`` functions by monkey-patching ``Session``
 
-    Parameters are the same as in :class:`CachedSession`. Additional parameters:
+    Parameters are the same as in :py:class:`CachedSession`. Additional parameters:
 
     Args:
         session_factory: Session class to use. It must inherit from either :py:class:`CachedSession`
             or :py:class:`CacheMixin`
     """
-    if backend:
-        backend = backends.create_backend(backend, cache_name, kwargs)
 
     class _ConfiguredCachedSession(session_factory):
         def __init__(self):
@@ -312,30 +357,18 @@ def clear():
         get_cache().clear()
 
 
-def remove_expired_responses():
-    """Removes expired responses from storage"""
-    if is_installed():
-        return requests.Session().remove_expired_responses()
-
-
-def _get_absolute_time(expire_after: Union[int, float, datetime, timedelta]) -> Optional[datetime]:
-    """Convert a time value to an absolute datetime, if it's not already"""
-    if isinstance(expire_after, datetime):
-        return expire_after
-    if expire_after is None:
-        return None
-    return datetime.now(timezone.utc) + _get_timedelta(expire_after)  # type: ignore
-
+def remove_expired_responses(expire_after: ExpirationTime = None):
+    """Remove expired responses from the cache, optionally with revalidation
 
-def _get_timedelta(expire_after: Union[int, float, timedelta] = None) -> Optional[timedelta]:
-    """Convert a time value to a timedelta, if it's not already"""
-    if expire_after is not None and not isinstance(expire_after, timedelta):
-        expire_after = timedelta(seconds=expire_after)
-    return expire_after
+    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=CachedSession):
-    requests.Session = requests.sessions.Session = session_factory
+def _patch_session_factory(session_factory: Type[OriginalSession] = CachedSession):
+    requests.Session = requests.sessions.Session = session_factory  # noqa
 
 
 def _normalize_parameters(params: Optional[Dict]) -> Dict:
diff --git a/requests_cache/response.py b/requests_cache/response.py
new file mode 100755
index 0000000..3eb3afa
--- /dev/null
+++ b/requests_cache/response.py
@@ -0,0 +1,156 @@
+from copy import copy
+from datetime import datetime, timedelta
+from io import BytesIO
+from typing import Any, Dict, Optional, Union
+
+from requests import Response
+from urllib3.response import HTTPResponse
+
+ExpirationTime = Union[None, int, float, datetime, timedelta]
+
+# Reponse attributes to copy
+RESPONSE_ATTRS = Response.__attrs__
+RAW_RESPONSE_ATTRS = [
+    'decode_content',
+    'headers',
+    'reason',
+    'request_method',
+    'request_url',
+    'status',
+    'strict',
+    'version',
+]
+
+
+class CachedResponse(Response):
+    """A serializable wrapper for :py:class:`requests.Response`. CachedResponse objects will behave
+    the same as the original response, but with some additional cache-related details. This class is
+    responsible for converting and setting cache expiration times, and converting response info into
+    a serializable format.
+
+    Args:
+        original_response: Response object
+        expire_after:
+    """
+
+    def __init__(self, original_response: Response, expire_after: ExpirationTime = None):
+        """Create a CachedResponse based on an original Response"""
+        super().__init__()
+        # Set cache-specific attrs
+        self.created_at = datetime.utcnow()
+        self.expires = self._get_expiration_datetime(expire_after)
+        self.from_cache = True
+
+        # Copy basic response attrs and original request
+        for k in RESPONSE_ATTRS:
+            setattr(self, k, getattr(original_response, k, None))
+        self.request = copy(original_response.request)
+        self.request.hooks = []
+
+        # Read content to support streaming requests, and reset file pointer on original request
+        self._content = original_response.content
+        original_response.raw._fp = BytesIO(self._content or b'')
+
+        # Copy raw response
+        self._raw_response = None
+        self._raw_response_attrs: Dict[str, Any] = {}
+        for k in RAW_RESPONSE_ATTRS:
+            self._raw_response_attrs[k] = getattr(original_response.raw, k, None)
+
+        # Copy redirect history, if any
+        self.history = []
+        for redirect in original_response.history:
+            self.history.append(CachedResponse(redirect))
+
+    def __getstate__(self):
+        """Override pickling behavior in ``requests.Response.__getstate__``"""
+        return self.__dict__
+
+    def _get_expiration_datetime(self, expire_after: ExpirationTime) -> Optional[datetime]:
+        """Convert a time value or delta to an absolute datetime, if it's not already"""
+        if expire_after is None or expire_after == -1:
+            return None
+        elif isinstance(expire_after, datetime):
+            return expire_after
+
+        if not isinstance(expire_after, timedelta):
+            expire_after = timedelta(seconds=expire_after)
+        return self.created_at + expire_after
+
+    def reset(self):
+        """Reset raw response file handler, if previously initialized"""
+        self._raw_response = None
+
+    @property
+    def is_expired(self) -> bool:
+        """Determine if this cached response is expired"""
+        return self.expires is not None and datetime.utcnow() > self.expires
+
+    @property
+    def raw(self) -> HTTPResponse:
+        """Reconstruct a raw urllib response object from stored attrs"""
+        if not self._raw_response:
+            self._raw_response = CachedHTTPResponse(body=self._content, **self._raw_response_attrs)
+        return self._raw_response
+
+    @raw.setter
+    def raw(self, value):
+        """No-op to handle requests.Response attempting to set self.raw"""
+
+    def revalidate(self, expire_after: ExpirationTime) -> bool:
+        """Set a new expiration for this response, and determine if it is now expired"""
+        self.expires = self._get_expiration_datetime(expire_after)
+        return self.is_expired
+
+
+class CachedHTTPResponse(HTTPResponse):
+    """A wrapper for raw urllib response objects, which wraps cached content with support for
+    streaming requests
+    """
+
+    def __init__(self, body: bytes = None, **kwargs):
+        kwargs.setdefault('preload_content', False)
+        super().__init__(body=BytesIO(body or b''), **kwargs)
+        self._body = body
+
+    def release_conn(self):
+        """No-op for compatibility"""
+
+    def read(self, amt=None, decode_content=False, **kwargs):
+        """Simplified reader for cached content that emulates
+        :py:meth:`urllib3.response.HTTPResponse.read()`
+        """
+        data = self._fp.read(amt)
+        decode_content = self.decode_content if decode_content is None else decode_content
+
+        # "close" the file to inform consumers to stop reading from it
+        if not data:
+            self._fp.close()
+        # Decode binary content, if specified
+        elif decode_content:
+            self._init_decoder()
+            data = self._decode(data, decode_content=True, flush_decoder=True)
+
+        return data
+
+    def stream(self, amt=None, **kwargs):
+        """Simplified generator over cached content that emulates
+        :py:meth:`urllib3.response.HTTPResponse.stream()`
+        """
+        while not self._fp.closed:
+            yield self.read(amt=amt, **kwargs)
+
+
+AnyResponse = Union[Response, CachedResponse]
+
+
+def set_response_defaults(response: AnyResponse) -> AnyResponse:
+    """Set some default CachedResponse values on a requests.Response object, so they can be
+    expected to always be present
+    """
+    if not isinstance(response, CachedResponse):
+        response.created_at = None
+        response.expires = None
+        response.from_cache = False
+        response.is_expired = False
+    return response