<s> """<STR_LIT>""" <EOL> __all__ = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> import collections <EOL> import os <EOL> import urlparse <EOL> from . import api_utils <EOL> from . import common <EOL> from . import errors <EOL> from . import rest_api <EOL> try : <EOL> from google . appengine . api import urlfetch <EOL> from google . appengine . ext import ndb <EOL> except ImportError : <EOL> from google . appengine . api import urlfetch <EOL> from google . appengine . ext import ndb <EOL> def _get_storage_api ( retry_params , account_id = None ) : <EOL> """<STR_LIT>""" <EOL> api = _StorageApi ( _StorageApi . full_control_scope , <EOL> service_account_id = account_id , <EOL> retry_params = retry_params ) <EOL> if common . local_run ( ) and not common . get_access_token ( ) : <EOL> api . api_url = common . local_api_url ( ) <EOL> if common . get_access_token ( ) : <EOL> api . token = common . get_access_token ( ) <EOL> return api <EOL> class _StorageApi ( rest_api . _RestApi ) : <EOL> """<STR_LIT>""" <EOL> api_url = '<STR_LIT>' <EOL> read_only_scope = '<STR_LIT>' <EOL> read_write_scope = '<STR_LIT>' <EOL> full_control_scope = '<STR_LIT>' <EOL> def __getstate__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( super ( _StorageApi , self ) . __getstate__ ( ) , { '<STR_LIT>' : self . api_url } ) <EOL> def __setstate__ ( self , state ) : <EOL> """<STR_LIT>""" <EOL> superstate , localstate = state <EOL> super ( _StorageApi , self ) . __setstate__ ( superstate ) <EOL> self . api_url = localstate [ '<STR_LIT>' ] <EOL> @ api_utils . _eager_tasklet <EOL> @ ndb . tasklet <EOL> def do_request_async ( self , url , method = '<STR_LIT:GET>' , headers = None , payload = None , <EOL> deadline = None , callback = None ) : <EOL> """<STR_LIT>""" <EOL> if headers is None : <EOL> headers = { } <EOL> if '<STR_LIT>' not in headers : <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT:2>' <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> try : <EOL> resp_tuple = yield super ( _StorageApi , self ) . do_request_async ( <EOL> url , method = method , headers = headers , payload = payload , <EOL> deadline = deadline , callback = callback ) <EOL> except urlfetch . DownloadError , e : <EOL> raise errors . TimeoutError ( <EOL> '<STR_LIT>' , e ) <EOL> raise ndb . Return ( resp_tuple ) <EOL> def post_object_async ( self , path , ** kwds ) : <EOL> """<STR_LIT>""" <EOL> return self . do_request_async ( self . api_url + path , '<STR_LIT:POST>' , ** kwds ) <EOL> def put_object_async ( self , path , ** kwds ) : <EOL> """<STR_LIT>""" <EOL> return self . do_request_async ( self . api_url + path , '<STR_LIT>' , ** kwds ) <EOL> def get_object_async ( self , path , ** kwds ) : <EOL> """<STR_LIT>""" <EOL> return self . do_request_async ( self . api_url + path , '<STR_LIT:GET>' , ** kwds ) <EOL> def delete_object_async ( self , path , ** kwds ) : <EOL> """<STR_LIT>""" <EOL> return self . do_request_async ( self . api_url + path , '<STR_LIT>' , ** kwds ) <EOL> def head_object_async ( self , path , ** kwds ) : <EOL> """<STR_LIT>""" <EOL> return self . do_request_async ( self . api_url + path , '<STR_LIT>' , ** kwds ) <EOL> def get_bucket_async ( self , path , ** kwds ) : <EOL> """<STR_LIT>""" <EOL> return self . do_request_async ( self . api_url + path , '<STR_LIT:GET>' , ** kwds ) <EOL> _StorageApi = rest_api . add_sync_methods ( _StorageApi ) <EOL> class ReadBuffer ( object ) : <EOL> """<STR_LIT>""" <EOL> DEFAULT_BUFFER_SIZE = <NUM_LIT> * <NUM_LIT> <EOL> MAX_REQUEST_SIZE = <NUM_LIT:30> * DEFAULT_BUFFER_SIZE <EOL> def __init__ ( self , <EOL> api , <EOL> path , <EOL> buffer_size = DEFAULT_BUFFER_SIZE , <EOL> max_request_size = MAX_REQUEST_SIZE ) : <EOL> """<STR_LIT>""" <EOL> self . _api = api <EOL> self . _path = path <EOL> self . name = api_utils . _unquote_filename ( path ) <EOL> self . closed = False <EOL> assert buffer_size <= max_request_size <EOL> self . _buffer_size = buffer_size <EOL> self . _max_request_size = max_request_size <EOL> self . _offset = <NUM_LIT:0> <EOL> self . _buffer = _Buffer ( ) <EOL> self . _etag = None <EOL> self . _request_next_buffer ( ) <EOL> status , headers , content = self . _api . head_object ( path ) <EOL> errors . check_status ( status , [ <NUM_LIT:200> ] , path , resp_headers = headers , body = content ) <EOL> self . _file_size = long ( headers [ '<STR_LIT>' ] ) <EOL> self . _check_etag ( headers . get ( '<STR_LIT>' ) ) <EOL> if self . _file_size == <NUM_LIT:0> : <EOL> self . _buffer_future = None <EOL> def __getstate__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : self . _api , <EOL> '<STR_LIT:path>' : self . _path , <EOL> '<STR_LIT>' : self . _buffer_size , <EOL> '<STR_LIT>' : self . _max_request_size , <EOL> '<STR_LIT>' : self . _etag , <EOL> '<STR_LIT:size>' : self . _file_size , <EOL> '<STR_LIT>' : self . _offset , <EOL> '<STR_LIT>' : self . closed } <EOL> def __setstate__ ( self , state ) : <EOL> """<STR_LIT>""" <EOL> self . _api = state [ '<STR_LIT>' ] <EOL> self . _path = state [ '<STR_LIT:path>' ] <EOL> self . name = api_utils . _unquote_filename ( self . _path ) <EOL> self . _buffer_size = state [ '<STR_LIT>' ] <EOL> self . _max_request_size = state [ '<STR_LIT>' ] <EOL> self . _etag = state [ '<STR_LIT>' ] <EOL> self . _file_size = state [ '<STR_LIT:size>' ] <EOL> self . _offset = state [ '<STR_LIT>' ] <EOL> self . _buffer = _Buffer ( ) <EOL> self . closed = state [ '<STR_LIT>' ] <EOL> self . _buffer_future = None <EOL> if self . _remaining ( ) and not self . closed : <EOL> self . _request_next_buffer ( ) <EOL> def __iter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self <EOL> def next ( self ) : <EOL> line = self . readline ( ) <EOL> if not line : <EOL> raise StopIteration ( ) <EOL> return line <EOL> def readline ( self , size = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> self . _check_open ( ) <EOL> if size == <NUM_LIT:0> or not self . _remaining ( ) : <EOL> return '<STR_LIT>' <EOL> data_list = [ ] <EOL> newline_offset = self . _buffer . find_newline ( size ) <EOL> while newline_offset < <NUM_LIT:0> : <EOL> data = self . _buffer . read ( size ) <EOL> size -= len ( data ) <EOL> self . _offset += len ( data ) <EOL> data_list . append ( data ) <EOL> if size == <NUM_LIT:0> or not self . _remaining ( ) : <EOL> return '<STR_LIT>' . join ( data_list ) <EOL> self . _buffer . reset ( self . _buffer_future . get_result ( ) ) <EOL> self . _request_next_buffer ( ) <EOL> newline_offset = self . _buffer . find_newline ( size ) <EOL> data = self . _buffer . read_to_offset ( newline_offset + <NUM_LIT:1> ) <EOL> self . _offset += len ( data ) <EOL> data_list . append ( data ) <EOL> return '<STR_LIT>' . join ( data_list ) <EOL> def read ( self , size = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> self . _check_open ( ) <EOL> if not self . _remaining ( ) : <EOL> return '<STR_LIT>' <EOL> data_list = [ ] <EOL> while True : <EOL> remaining = self . _buffer . remaining ( ) <EOL> if size >= <NUM_LIT:0> and size < remaining : <EOL> data_list . append ( self . _buffer . read ( size ) ) <EOL> self . _offset += size <EOL> break <EOL> else : <EOL> size -= remaining <EOL> self . _offset += remaining <EOL> data_list . append ( self . _buffer . read ( ) ) <EOL> if self . _buffer_future is None : <EOL> if size < <NUM_LIT:0> or size >= self . _remaining ( ) : <EOL> needs = self . _remaining ( ) <EOL> else : <EOL> needs = size <EOL> data_list . extend ( self . _get_segments ( self . _offset , needs ) ) <EOL> self . _offset += needs <EOL> break <EOL> if self . _buffer_future : <EOL> self . _buffer . reset ( self . _buffer_future . get_result ( ) ) <EOL> self . _buffer_future = None <EOL> if self . _buffer_future is None : <EOL> self . _request_next_buffer ( ) <EOL> return '<STR_LIT>' . join ( data_list ) <EOL> def _remaining ( self ) : <EOL> return self . _file_size - self . _offset <EOL> def _request_next_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _buffer_future = None <EOL> next_offset = self . _offset + self . _buffer . remaining ( ) <EOL> if not hasattr ( self , '<STR_LIT>' ) or next_offset != self . _file_size : <EOL> self . _buffer_future = self . _get_segment ( next_offset , <EOL> self . _buffer_size ) <EOL> def _get_segments ( self , start , request_size ) : <EOL> """<STR_LIT>""" <EOL> if not request_size : <EOL> return [ ] <EOL> end = start + request_size <EOL> futures = [ ] <EOL> while request_size > self . _max_request_size : <EOL> futures . append ( self . _get_segment ( start , self . _max_request_size ) ) <EOL> request_size -= self . _max_request_size <EOL> start += self . _max_request_size <EOL> if start < end : <EOL> futures . append ( self . _get_segment ( start , end - start ) ) <EOL> return [ fut . get_result ( ) for fut in futures ] <EOL> @ ndb . tasklet <EOL> def _get_segment ( self , start , request_size ) : <EOL> """<STR_LIT>""" <EOL> end = start + request_size - <NUM_LIT:1> <EOL> content_range = '<STR_LIT>' % ( start , end ) <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' + content_range } <EOL> status , resp_headers , content = yield self . _api . get_object_async ( <EOL> self . _path , headers = headers ) <EOL> errors . check_status ( status , [ <NUM_LIT:200> , <NUM_LIT> ] , self . _path , headers , resp_headers , <EOL> body = content ) <EOL> self . _check_etag ( resp_headers . get ( '<STR_LIT>' ) ) <EOL> raise ndb . Return ( content ) <EOL> def _check_etag ( self , etag ) : <EOL> """<STR_LIT>""" <EOL> if etag is None : <EOL> return <EOL> elif self . _etag is None : <EOL> self . _etag = etag <EOL> elif self . _etag != etag : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> def close ( self ) : <EOL> self . closed = True <EOL> self . _buffer = None <EOL> self . _buffer_future = None <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , atype , value , traceback ) : <EOL> self . close ( ) <EOL> return False <EOL> def seek ( self , offset , whence = os . SEEK_SET ) : <EOL> """<STR_LIT>""" <EOL> self . _check_open ( ) <EOL> self . _buffer . reset ( ) <EOL> self . _buffer_future = None <EOL> if whence == os . SEEK_SET : <EOL> self . _offset = offset <EOL> elif whence == os . SEEK_CUR : <EOL> self . _offset += offset <EOL> elif whence == os . SEEK_END : <EOL> self . _offset = self . _file_size + offset <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % str ( whence ) ) <EOL> self . _offset = min ( self . _offset , self . _file_size ) <EOL> self . _offset = max ( self . _offset , <NUM_LIT:0> ) <EOL> if self . _remaining ( ) : <EOL> self . _request_next_buffer ( ) <EOL> def tell ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _check_open ( ) <EOL> return self . _offset <EOL> def _check_open ( self ) : <EOL> if self . closed : <EOL> raise IOError ( '<STR_LIT>' ) <EOL> def seekable ( self ) : <EOL> return True <EOL> def readable ( self ) : <EOL> return True <EOL> def writable ( self ) : <EOL> return False <EOL> class _Buffer ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . reset ( ) <EOL> def reset ( self , content = '<STR_LIT>' , offset = <NUM_LIT:0> ) : <EOL> self . _buffer = content <EOL> self . _offset = offset <EOL> def read ( self , size = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> if size < <NUM_LIT:0> : <EOL> offset = len ( self . _buffer ) <EOL> else : <EOL> offset = self . _offset + size <EOL> return self . read_to_offset ( offset ) <EOL> def read_to_offset ( self , offset ) : <EOL> """<STR_LIT>""" <EOL> assert offset >= self . _offset <EOL> result = self . _buffer [ self . _offset : offset ] <EOL> self . _offset += len ( result ) <EOL> return result <EOL> def remaining ( self ) : <EOL> return len ( self . _buffer ) - self . _offset <EOL> def find_newline ( self , size = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> if size < <NUM_LIT:0> : <EOL> return self . _buffer . find ( '<STR_LIT:\n>' , self . _offset ) <EOL> return self . _buffer . find ( '<STR_LIT:\n>' , self . _offset , self . _offset + size ) <EOL> class StreamingBuffer ( object ) : <EOL> """<STR_LIT>""" <EOL> _blocksize = <NUM_LIT> * <NUM_LIT> <EOL> _flushsize = <NUM_LIT:8> * _blocksize <EOL> _maxrequestsize = <NUM_LIT:9> * <NUM_LIT:4> * _blocksize <EOL> def __init__ ( self , <EOL> api , <EOL> path , <EOL> content_type = None , <EOL> gcs_headers = None ) : <EOL> """<STR_LIT>""" <EOL> assert self . _maxrequestsize > self . _blocksize <EOL> assert self . _maxrequestsize % self . _blocksize == <NUM_LIT:0> <EOL> assert self . _maxrequestsize >= self . _flushsize <EOL> self . _api = api <EOL> self . _path = path <EOL> self . name = api_utils . _unquote_filename ( path ) <EOL> self . closed = False <EOL> self . _buffer = collections . deque ( ) <EOL> self . _buffered = <NUM_LIT:0> <EOL> self . _written = <NUM_LIT:0> <EOL> self . _offset = <NUM_LIT:0> <EOL> headers = { '<STR_LIT>' : '<STR_LIT:start>' } <EOL> if content_type : <EOL> headers [ '<STR_LIT>' ] = content_type <EOL> if gcs_headers : <EOL> headers . update ( gcs_headers ) <EOL> status , resp_headers , content = self . _api . post_object ( path , headers = headers ) <EOL> errors . check_status ( status , [ <NUM_LIT> ] , path , headers , resp_headers , <EOL> body = content ) <EOL> loc = resp_headers . get ( '<STR_LIT:location>' ) <EOL> if not loc : <EOL> raise IOError ( '<STR_LIT>' ) <EOL> parsed = urlparse . urlparse ( loc ) <EOL> self . _path_with_token = '<STR_LIT>' % ( self . _path , parsed . query ) <EOL> def __getstate__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return { '<STR_LIT>' : self . _api , <EOL> '<STR_LIT:path>' : self . _path , <EOL> '<STR_LIT>' : self . _path_with_token , <EOL> '<STR_LIT>' : self . _buffer , <EOL> '<STR_LIT>' : self . _buffered , <EOL> '<STR_LIT>' : self . _written , <EOL> '<STR_LIT>' : self . _offset , <EOL> '<STR_LIT>' : self . closed } <EOL> def __setstate__ ( self , state ) : <EOL> """<STR_LIT>""" <EOL> self . _api = state [ '<STR_LIT>' ] <EOL> self . _path_with_token = state [ '<STR_LIT>' ] <EOL> self . _buffer = state [ '<STR_LIT>' ] <EOL> self . _buffered = state [ '<STR_LIT>' ] <EOL> self . _written = state [ '<STR_LIT>' ] <EOL> self . _offset = state [ '<STR_LIT>' ] <EOL> self . closed = state [ '<STR_LIT>' ] <EOL> self . _path = state [ '<STR_LIT:path>' ] <EOL> self . name = api_utils . _unquote_filename ( self . _path ) <EOL> def write ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . _check_open ( ) <EOL> if not isinstance ( data , str ) : <EOL> raise TypeError ( '<STR_LIT>' % type ( data ) ) <EOL> if not data : <EOL> return <EOL> self . _buffer . append ( data ) <EOL> self . _buffered += len ( data ) <EOL> self . _offset += len ( data ) <EOL> if self . _buffered >= self . _flushsize : <EOL> self . _flush ( ) <EOL> def flush ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _check_open ( ) <EOL> self . _flush ( finish = False ) <EOL> def tell ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _offset <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . closed : <EOL> self . closed = True <EOL> self . _flush ( finish = True ) <EOL> self . _buffer = None <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , atype , value , traceback ) : <EOL> self . close ( ) <EOL> return False <EOL> def _flush ( self , finish = False ) : <EOL> """<STR_LIT>""" <EOL> blocksize_or_zero = <NUM_LIT:0> if finish else self . _blocksize <EOL> while self . _buffered >= blocksize_or_zero : <EOL> buffer = [ ] <EOL> buffered = <NUM_LIT:0> <EOL> excess = <NUM_LIT:0> <EOL> while self . _buffer : <EOL> buf = self . _buffer . popleft ( ) <EOL> size = len ( buf ) <EOL> self . _buffered -= size <EOL> buffer . append ( buf ) <EOL> buffered += size <EOL> if buffered >= self . _maxrequestsize : <EOL> excess = buffered - self . _maxrequestsize <EOL> break <EOL> if self . _buffered < blocksize_or_zero and buffered >= blocksize_or_zero : <EOL> excess = buffered % self . _blocksize <EOL> break <EOL> if excess : <EOL> over = buffer . pop ( ) <EOL> size = len ( over ) <EOL> assert size >= excess <EOL> buffered -= size <EOL> head , tail = over [ : - excess ] , over [ - excess : ] <EOL> self . _buffer . appendleft ( tail ) <EOL> self . _buffered += len ( tail ) <EOL> if head : <EOL> buffer . append ( head ) <EOL> buffered += len ( head ) <EOL> data = '<STR_LIT>' . join ( buffer ) <EOL> file_len = '<STR_LIT:*>' <EOL> if finish and not self . _buffered : <EOL> file_len = self . _written + len ( data ) <EOL> self . _send_data ( data , self . _written , file_len ) <EOL> self . _written += len ( data ) <EOL> if file_len != '<STR_LIT:*>' : <EOL> break <EOL> def _send_data ( self , data , start_offset , file_len ) : <EOL> """<STR_LIT>""" <EOL> headers = { } <EOL> end_offset = start_offset + len ( data ) - <NUM_LIT:1> <EOL> if data : <EOL> headers [ '<STR_LIT>' ] = ( '<STR_LIT>' % <EOL> ( start_offset , end_offset , file_len ) ) <EOL> else : <EOL> headers [ '<STR_LIT>' ] = ( '<STR_LIT>' % file_len ) <EOL> status , response_headers , content = self . _api . put_object ( <EOL> self . _path_with_token , payload = data , headers = headers ) <EOL> if file_len == '<STR_LIT:*>' : <EOL> expected = <NUM_LIT> <EOL> else : <EOL> expected = <NUM_LIT:200> <EOL> errors . check_status ( status , [ expected ] , self . _path , headers , <EOL> response_headers , content , <EOL> { '<STR_LIT>' : self . _path_with_token } ) <EOL> def _get_offset_from_gcs ( self ) : <EOL> """<STR_LIT>""" <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> status , response_headers , content = self . _api . put_object ( <EOL> self . _path_with_token , headers = headers ) <EOL> errors . check_status ( status , [ <NUM_LIT> ] , self . _path , headers , <EOL> response_headers , content , <EOL> { '<STR_LIT>' : self . _path_with_token } ) <EOL> val = response_headers . get ( '<STR_LIT>' ) <EOL> if val is None : <EOL> return - <NUM_LIT:1> <EOL> _ , offset = val . rsplit ( '<STR_LIT:->' , <NUM_LIT:1> ) <EOL> return int ( offset ) <EOL> def _force_close ( self , file_length = None ) : <EOL> """<STR_LIT>""" <EOL> if file_length is None : <EOL> file_length = self . _get_offset_from_gcs ( ) + <NUM_LIT:1> <EOL> self . _send_data ( '<STR_LIT>' , <NUM_LIT:0> , file_length ) <EOL> def _check_open ( self ) : <EOL> if self . closed : <EOL> raise IOError ( '<STR_LIT>' ) <EOL> def seekable ( self ) : <EOL> return False <EOL> def readable ( self ) : <EOL> return False <EOL> def writable ( self ) : <EOL> return True </s>
