<s> """<STR_LIT>""" <EOL> import re <EOL> import calendar <EOL> import time <EOL> from email . utils import parsedate_tz <EOL> from pip . _vendor . requests . structures import CaseInsensitiveDict <EOL> from . cache import DictCache <EOL> from . serialize import Serializer <EOL> URI = re . compile ( r"<STR_LIT>" ) <EOL> def parse_uri ( uri ) : <EOL> """<STR_LIT>""" <EOL> groups = URI . match ( uri ) . groups ( ) <EOL> return ( groups [ <NUM_LIT:1> ] , groups [ <NUM_LIT:3> ] , groups [ <NUM_LIT:4> ] , groups [ <NUM_LIT:6> ] , groups [ <NUM_LIT:8> ] ) <EOL> class CacheController ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , cache = None , cache_etags = True , serializer = None ) : <EOL> self . cache = cache or DictCache ( ) <EOL> self . cache_etags = cache_etags <EOL> self . serializer = serializer or Serializer ( ) <EOL> @ classmethod <EOL> def _urlnorm ( cls , uri ) : <EOL> """<STR_LIT>""" <EOL> ( scheme , authority , path , query , fragment ) = parse_uri ( uri ) <EOL> if not scheme or not authority : <EOL> raise Exception ( "<STR_LIT>" % uri ) <EOL> scheme = scheme . lower ( ) <EOL> authority = authority . lower ( ) <EOL> if not path : <EOL> path = "<STR_LIT:/>" <EOL> request_uri = query and "<STR_LIT:?>" . join ( [ path , query ] ) or path <EOL> defrag_uri = scheme + "<STR_LIT>" + authority + request_uri <EOL> return defrag_uri <EOL> @ classmethod <EOL> def cache_url ( cls , uri ) : <EOL> return cls . _urlnorm ( uri ) <EOL> def parse_cache_control ( self , headers ) : <EOL> """<STR_LIT>""" <EOL> retval = { } <EOL> cc_header = '<STR_LIT>' <EOL> if '<STR_LIT>' in headers : <EOL> cc_header = '<STR_LIT>' <EOL> if cc_header in headers : <EOL> parts = headers [ cc_header ] . split ( '<STR_LIT:U+002C>' ) <EOL> parts_with_args = [ <EOL> tuple ( [ x . strip ( ) . lower ( ) for x in part . split ( "<STR_LIT:=>" , <NUM_LIT:1> ) ] ) <EOL> for part in parts if - <NUM_LIT:1> != part . find ( "<STR_LIT:=>" ) <EOL> ] <EOL> parts_wo_args = [ <EOL> ( name . strip ( ) . lower ( ) , <NUM_LIT:1> ) <EOL> for name in parts if - <NUM_LIT:1> == name . find ( "<STR_LIT:=>" ) <EOL> ] <EOL> retval = dict ( parts_with_args + parts_wo_args ) <EOL> return retval <EOL> def cached_request ( self , request ) : <EOL> """<STR_LIT>""" <EOL> cache_url = self . cache_url ( request . url ) <EOL> cc = self . parse_cache_control ( request . headers ) <EOL> no_cache = True if '<STR_LIT>' in cc else False <EOL> if '<STR_LIT>' in cc and cc [ '<STR_LIT>' ] == <NUM_LIT:0> : <EOL> no_cache = True <EOL> if no_cache : <EOL> return False <EOL> resp = self . serializer . loads ( request , self . cache . get ( cache_url ) ) <EOL> if not resp : <EOL> return False <EOL> if resp . status == <NUM_LIT> : <EOL> return resp <EOL> headers = CaseInsensitiveDict ( resp . headers ) <EOL> if not headers or '<STR_LIT:date>' not in headers : <EOL> if '<STR_LIT>' not in headers : <EOL> self . cache . delete ( cache_url ) <EOL> return False <EOL> now = time . time ( ) <EOL> date = calendar . timegm ( <EOL> parsedate_tz ( headers [ '<STR_LIT:date>' ] ) <EOL> ) <EOL> current_age = max ( <NUM_LIT:0> , now - date ) <EOL> resp_cc = self . parse_cache_control ( headers ) <EOL> freshness_lifetime = <NUM_LIT:0> <EOL> if '<STR_LIT>' in resp_cc and resp_cc [ '<STR_LIT>' ] . isdigit ( ) : <EOL> freshness_lifetime = int ( resp_cc [ '<STR_LIT>' ] ) <EOL> elif '<STR_LIT>' in headers : <EOL> expires = parsedate_tz ( headers [ '<STR_LIT>' ] ) <EOL> if expires is not None : <EOL> expire_time = calendar . timegm ( expires ) - date <EOL> freshness_lifetime = max ( <NUM_LIT:0> , expire_time ) <EOL> if '<STR_LIT>' in cc : <EOL> try : <EOL> freshness_lifetime = int ( cc [ '<STR_LIT>' ] ) <EOL> except ValueError : <EOL> freshness_lifetime = <NUM_LIT:0> <EOL> if '<STR_LIT>' in cc : <EOL> try : <EOL> min_fresh = int ( cc [ '<STR_LIT>' ] ) <EOL> except ValueError : <EOL> min_fresh = <NUM_LIT:0> <EOL> current_age += min_fresh <EOL> fresh = ( freshness_lifetime > current_age ) <EOL> if fresh : <EOL> return resp <EOL> if '<STR_LIT>' not in headers : <EOL> self . cache . delete ( cache_url ) <EOL> return False <EOL> def conditional_headers ( self , request ) : <EOL> cache_url = self . cache_url ( request . url ) <EOL> resp = self . serializer . loads ( request , self . cache . get ( cache_url ) ) <EOL> new_headers = { } <EOL> if resp : <EOL> headers = CaseInsensitiveDict ( resp . headers ) <EOL> if '<STR_LIT>' in headers : <EOL> new_headers [ '<STR_LIT>' ] = headers [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in headers : <EOL> new_headers [ '<STR_LIT>' ] = headers [ '<STR_LIT>' ] <EOL> return new_headers <EOL> def cache_response ( self , request , response , body = None ) : <EOL> """<STR_LIT>""" <EOL> if response . status not in [ <NUM_LIT:200> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] : <EOL> return <EOL> response_headers = CaseInsensitiveDict ( response . headers ) <EOL> cc_req = self . parse_cache_control ( request . headers ) <EOL> cc = self . parse_cache_control ( response_headers ) <EOL> cache_url = self . cache_url ( request . url ) <EOL> no_store = cc . get ( '<STR_LIT>' ) or cc_req . get ( '<STR_LIT>' ) <EOL> if no_store and self . cache . get ( cache_url ) : <EOL> self . cache . delete ( cache_url ) <EOL> if self . cache_etags and '<STR_LIT>' in response_headers : <EOL> self . cache . set ( <EOL> cache_url , <EOL> self . serializer . dumps ( request , response , body = body ) , <EOL> ) <EOL> elif response . status == <NUM_LIT> : <EOL> self . cache . set ( <EOL> cache_url , <EOL> self . serializer . dumps ( request , response ) <EOL> ) <EOL> elif '<STR_LIT:date>' in response_headers : <EOL> if cc and cc . get ( '<STR_LIT>' ) : <EOL> if int ( cc [ '<STR_LIT>' ] ) > <NUM_LIT:0> : <EOL> self . cache . set ( <EOL> cache_url , <EOL> self . serializer . dumps ( request , response , body = body ) , <EOL> ) <EOL> elif '<STR_LIT>' in response_headers : <EOL> if response_headers [ '<STR_LIT>' ] : <EOL> self . cache . set ( <EOL> cache_url , <EOL> self . serializer . dumps ( request , response , body = body ) , <EOL> ) <EOL> def update_cached_response ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> cache_url = self . cache_url ( request . url ) <EOL> cached_response = self . serializer . loads ( <EOL> request , <EOL> self . cache . get ( cache_url ) <EOL> ) <EOL> if not cached_response : <EOL> return response <EOL> excluded_headers = [ <EOL> "<STR_LIT>" , <EOL> ] <EOL> cached_response . headers . update ( <EOL> dict ( ( k , v ) for k , v in response . headers . items ( ) <EOL> if k . lower ( ) not in excluded_headers ) <EOL> ) <EOL> cached_response . status = <NUM_LIT:200> <EOL> self . cache . set ( <EOL> cache_url , <EOL> self . serializer . dumps ( request , cached_response ) , <EOL> ) <EOL> return cached_response </s>
