from datetime import datetime, timedelta, timezone
from logging import getLogger
from typing import TYPE_CHECKING, List, Optional, Tuple, Union
from attr import define, field
from requests import PreparedRequest, Response
from requests.cookies import RequestsCookieJar
from requests.structures import CaseInsensitiveDict
from urllib3.response import HTTPHeaderDict
from ..cache_control import ExpirationTime, get_expiration_datetime
from . import CachedHTTPResponse, CachedRequest
DATETIME_FORMAT = '%Y-%m-%d %H:%M:%S %Z' # Format used for __str__ only
HeaderList = List[Tuple[str, str]]
logger = getLogger(__name__)
[docs]@define(auto_attribs=False, slots=False)
class CachedResponse(Response):
"""A class that emulates :py:class:`requests.Response`, with some additional optimizations
for serialization.
"""
_content: bytes = field(default=None)
_next: Optional[CachedRequest] = field(default=None)
cache_key: Optional[str] = None # Not serialized; set by BaseCache.get_response()
cookies: RequestsCookieJar = field(factory=RequestsCookieJar)
created_at: datetime = field(factory=datetime.utcnow)
elapsed: timedelta = field(factory=timedelta)
encoding: str = field(default=None)
expires: Optional[datetime] = field(default=None)
headers: CaseInsensitiveDict = field(factory=CaseInsensitiveDict)
history: List['CachedResponse'] = field(factory=list) # type: ignore
raw: CachedHTTPResponse = field(factory=CachedHTTPResponse, repr=False)
reason: str = field(default=None)
request: CachedRequest = field(factory=CachedRequest) # type: ignore
status_code: int = field(default=0)
url: str = field(default=None)
def __attrs_post_init__(self):
"""Re-initialize raw response body after deserialization"""
if self.raw._body is None and self._content is not None:
self.raw.reset(self._content)
if not self.raw.headers:
self.raw.headers = HTTPHeaderDict(self.headers)
[docs] @classmethod
def from_response(cls, original_response: Response, **kwargs):
"""Create a CachedResponse based on an original response object"""
obj = cls(**kwargs)
# Copy basic attributes
for k in Response.__attrs__:
setattr(obj, k, getattr(original_response, k, None))
# Store request, raw response, and next response (if it's a redirect response)
obj.request = CachedRequest.from_request(original_response.request)
obj.raw = CachedHTTPResponse.from_response(original_response)
obj._next = (
CachedRequest.from_request(original_response.next) if original_response.next else None
)
# Store response body, which will have been read & decoded by requests.Response by now
obj._content = original_response.content
# Copy redirect history, if any; avoid recursion by not copying redirects of redirects
obj.history = []
if not obj.is_redirect:
for redirect in original_response.history:
obj.history.append(cls.from_response(redirect))
return obj
@property
def _content_consumed(self) -> bool:
"""For compatibility with requests.Response; will always be True for a cached response"""
return True
@_content_consumed.setter
def _content_consumed(self, value: bool):
pass
@property
def from_cache(self) -> bool:
return True
@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 next(self) -> Optional[PreparedRequest]:
"""Returns a PreparedRequest for the next request in a redirect chain, if there is one."""
return self._next.prepare() if self._next else None
[docs] def revalidate(self, expire_after: ExpirationTime) -> bool:
"""Set a new expiration for this response, and determine if it is now expired"""
self.expires = get_expiration_datetime(expire_after)
return self.is_expired
[docs] def reset(self):
if self.raw:
self.raw.reset()
@property
def size(self) -> int:
"""Get the size of the response body in bytes"""
return len(self.content) if self.content else 0
def __str__(self):
return (
f'request: {self.request}, response: {self.status_code} '
f'({format_file_size(self.size)}), created: {format_datetime(self.created_at)}, '
f'expires: {format_datetime(self.expires)} ({"stale" if self.is_expired else "fresh"})'
)
[docs]def set_response_defaults(
response: Union[Response, CachedResponse], cache_key: str = None
) -> Union[Response, CachedResponse]:
"""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.cache_key = cache_key # type: ignore
response.created_at = None # type: ignore
response.expires = None # type: ignore
response.from_cache = False # type: ignore
response.is_expired = False # type: ignore
return response