from__future__importannotationsfromdatetimeimportdatetime,timedelta,timezonefromloggingimportgetLoggerfromtimeimporttimefromtypingimportTYPE_CHECKING,Dict,List,Optional,Unionimportattrfromattrimportdefine,fieldfromrequestsimportPreparedRequest,Responsefromrequests.cookiesimportRequestsCookieJarfromrequests.structuresimportCaseInsensitiveDictfrom..policyimportExpirationTime,get_expiration_datetimefrom.importCachedHTTPResponse,CachedRequest,RichMixinifTYPE_CHECKING:from..policy.actionsimportCacheActionsDATETIME_FORMAT='%Y-%m-%d %H:%M:%S %Z'# Format used for __str__ onlyDecodedContent=Union[Dict,List,str,None]logger=getLogger(__name__)
[docs]@define(auto_attribs=False,repr=False,slots=False)classBaseResponse(Response):"""Wrapper class for responses returned by :py:class:`.CachedSession`. This mainly exists to provide type hints for extra cache-related attributes that are added to non-cached responses. """created_at:datetime=field(factory=datetime.utcnow)expires:Optional[datetime]=field(default=None)cache_key:str=''# Not serialized; set by BaseCache.get_response()revalidated:bool=False# Not serialized; set by CacheActions.update_revalidated_response()@propertydeffrom_cache(self)->bool:returnFalse@propertydefis_expired(self)->bool:returnFalse
[docs]@define(auto_attribs=False,repr=False,slots=False)classOriginalResponse(BaseResponse):"""Wrapper class for non-cached responses returned by :py:class:`.CachedSession`"""
[docs]@classmethoddefwrap_response(cls,response:Response,actions:'CacheActions')->'OriginalResponse':"""Modify a response object in-place and add extra cache-related attributes"""ifnotisinstance(response,cls):response.__class__=cls# Add expires and cache_key only if the response was written to the cacheresponse.expires=Noneifactions.skip_writeelseactions.expires# type: ignoreresponse.cache_key=Noneifactions.skip_writeelseactions.cache_key# type: ignoreresponse.created_at=datetime.utcnow()# type: ignorereturnresponse# type: ignore
[docs]@define(auto_attribs=False,repr=False,slots=False)classCachedResponse(RichMixin,BaseResponse):"""A class that emulates :py:class:`requests.Response`, optimized for serialization"""_content:bytes=field(default=None)_decoded_content:DecodedContent=field(default=None)_next:Optional[CachedRequest]=field(default=None)cookies:RequestsCookieJar=field(factory=RequestsCookieJar)created_at:datetime=field(default=None)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: ignoreraw:CachedHTTPResponse=None# type: ignore # Not serialized; populated from CachedResponse attrsreason:str=field(default=None)request:CachedRequest=field(factory=CachedRequest)# type: ignorestatus_code:int=field(default=0)url:str=field(default=None)def__attrs_post_init__(self):# Not using created_at field default due to possible bug on Windows with omit_if_defaultself.created_at=self.created_atordatetime.utcnow()# Re-initialize raw (urllib3) response after deserializationself.raw=self.raworCachedHTTPResponse.from_cached_response(self)
[docs]@classmethoddeffrom_response(cls,response:Response,**kwargs)->'CachedResponse':"""Create a CachedResponse based on an original Response or another CachedResponse object"""ifisinstance(response,CachedResponse):obj=attr.evolve(response,**kwargs)obj._convert_redirects()returnobjobj=cls(**kwargs)# Copy basic attributesforkinResponse.__attrs__:setattr(obj,k,getattr(response,k,None))# Store request, raw response, and next response (if it's a redirect response)obj.raw=CachedHTTPResponse.from_response(response)obj.request=CachedRequest.from_request(response.request)obj._next=CachedRequest.from_request(response.next)ifresponse.nextelseNone# Store response body, which will have been read & decoded by requests.Response by nowobj._content=response.contentobj._convert_redirects()returnobj
def_convert_redirects(self):"""Convert redirect history, if any; avoid recursion by not copying redirects of redirects"""ifself.is_redirect:self.history=[]returnself.history=[self.from_response(redirect)forredirectinself.history]@propertydef_content_consumed(self)->bool:"""For compatibility with requests.Response; will always be True for a cached response"""returnTrue@_content_consumed.setterdef_content_consumed(self,value:bool):pass@propertydefexpires_delta(self)->Optional[int]:"""Get time to expiration in seconds (rounded to the nearest second)"""ifself.expiresisNone:returnNonedelta=self.expires-datetime.utcnow()returnround(delta.total_seconds())@propertydefexpires_unix(self)->Optional[int]:"""Get expiration time as a Unix timestamp"""seconds=self.expires_deltareturnround(time()+seconds)ifsecondsisnotNoneelseNone@propertydeffrom_cache(self)->bool:returnTrue@propertydefis_expired(self)->bool:"""Determine if this cached response is expired"""returnself.expiresisnotNoneanddatetime.utcnow()>=self.expires
[docs]defis_older_than(self,older_than:ExpirationTime)->bool:"""Determine if this cached response is older than the given time"""older_than=get_expiration_datetime(older_than,negative_delta=True)returnolder_thanisnotNoneandself.created_at<older_than
@propertydefnext(self)->Optional[PreparedRequest]:"""Returns a PreparedRequest for the next request in a redirect chain, if there is one."""returnself._next.prepare()ifself._nextelseNone
[docs]defreset_expiration(self,expire_after:ExpirationTime):"""Set a new expiration for this response"""self.expires=get_expiration_datetime(expire_after)returnself.is_expired
@propertydefsize(self)->int:"""Get the size of the response body in bytes"""returnlen(self.content)ifself.contentelse0def__getstate__(self):"""Override pickling behavior from ``requests.Response.__getstate__``"""returnself.__dict__def__setstate__(self,state):"""Override pickling behavior from ``requests.Response.__setstate__``"""forname,valueinstate.items():setattr(self,name,value)def__str__(self):return(f'<CachedResponse [{self.status_code}]: 'f'created: {format_datetime(self.created_at)}, 'f'expires: {format_datetime(self.expires)} ({"stale"ifself.is_expiredelse"fresh"}), 'f'size: {format_file_size(self.size)}, request: {self.request}>')
[docs]defformat_datetime(value:Optional[datetime])->str:"""Get a formatted datetime string in the local time zone"""ifnotvalue:return"N/A"ifvalue.tzinfoisNone:value=value.replace(tzinfo=timezone.utc)returnvalue.astimezone().strftime(DATETIME_FORMAT)
[docs]defformat_file_size(n_bytes:int)->str:"""Convert a file size in bytes into a human-readable format"""filesize=float(n_bytesor0)def_format(unit):returnf'{int(filesize)}{unit}'ifunit=='bytes'elsef'{filesize:.2f}{unit}'forunitin['bytes','KiB','MiB','GiB']:iffilesize<1024orunit=='GiB':return_format(unit)filesize/=1024ifTYPE_CHECKING:return_format(unit)