#NEWLINE# def put_stream( self , bucket , label , fp , metadata = { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } , cb = None , num_cb = None ) : #NEWLINE# #INDENT# if ( metadata is None ) : #NEWLINE# #INDENT# metadata = { #NEWLINE# #INDENT# '_owner' : getpass . getuser ( ) , #NEWLINE# #UNINDENT# } #NEWLINE# #UNINDENT# path = ( ( ( '/' + bucket ) + '/' ) + label ) #NEWLINE# content_type = metadata . get ( '_format' , 'application/octet-stream' ) #NEWLINE# metadata = self . ckan . storage_metadata_set ( path , metadata ) #NEWLINE# BufferSize = 65536 #NEWLINE# headers = { #NEWLINE# #INDENT# 'Content-Type' : content_type , #NEWLINE# #UNINDENT# } #NEWLINE# m = md5 ( ) #NEWLINE# fp . seek ( 0 ) #NEWLINE# s = fp . read ( BufferSize ) #NEWLINE# while s : #NEWLINE# #INDENT# m . update ( s ) #NEWLINE# s = fp . read ( BufferSize ) #NEWLINE# #UNINDENT# self . size = fp . tell ( ) #NEWLINE# fp . seek ( 0 ) #NEWLINE# self . md5 = m . hexdigest ( ) #NEWLINE# headers [ 'Content-MD5' ] = base64 . encodestring ( m . digest ( ) ) . rstrip ( '\n' ) #NEWLINE# headers [ 'Content-Length' ] = str ( self . size ) #NEWLINE# headers [ 'Expect' ] = '100-Continue' #NEWLINE# ( host , headers ) = self . ckan . storage_auth_get ( path , headers ) #NEWLINE# def sender( http_conn , method , path , data , headers ) : #NEWLINE# #INDENT# http_conn . putrequest ( method , path ) #NEWLINE# for key in headers : #NEWLINE# #INDENT# http_conn . putheader ( key , headers [ key ] ) #NEWLINE# #UNINDENT# http_conn . endheaders ( ) #NEWLINE# fp . seek ( 0 ) #NEWLINE# http_conn . set_debuglevel ( 0 ) #NEWLINE# if cb : #NEWLINE# #INDENT# if ( num_cb > 2 ) : #NEWLINE# #INDENT# cb_count = ( ( self . size / BufferSize ) / ( num_cb - 2 ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( num_cb < 0 ) : #NEWLINE# #INDENT# cb_count = ( - 1 ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# cb_count = 0 #NEWLINE# #UNINDENT# #UNINDENT# i = total_bytes = 0 #NEWLINE# cb ( total_bytes , self . size ) #NEWLINE# #UNINDENT# l = fp . read ( BufferSize ) #NEWLINE# while ( len ( l ) > 0 ) : #NEWLINE# #INDENT# http_conn . send ( l ) #NEWLINE# if cb : #NEWLINE# #INDENT# total_bytes += len ( l ) #NEWLINE# i += 1 #NEWLINE# if ( ( i == cb_count ) or ( cb_count == ( - 1 ) ) ) : #NEWLINE# #INDENT# cb ( total_bytes , self . size ) #NEWLINE# i = 0 #NEWLINE# #UNINDENT# #UNINDENT# l = fp . read ( BufferSize ) #NEWLINE# #UNINDENT# if cb : #NEWLINE# #INDENT# cb ( total_bytes , self . size ) #NEWLINE# #UNINDENT# response = http_conn . getresponse ( ) #NEWLINE# body = response . read ( ) #NEWLINE# fp . seek ( 0 ) #NEWLINE# if ( ( response . status == 500 ) or ( response . status == 503 ) or response . getheader ( 'location' ) ) : #NEWLINE# #INDENT# return response #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( ( response . status >= 200 ) and ( response . status <= 299 ) ) : #NEWLINE# #INDENT# self . etag = response . getheader ( 'etag' ) #NEWLINE# if ( self . etag != ( '"%s"' % self . md5 ) ) : #NEWLINE# #INDENT# raise Exception ( 'ETag from S3 did not match computed MD5' ) #NEWLINE# #UNINDENT# return response #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# raise Exception ( response . status , response . reason , body ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# awsc = AWSAuthConnection ( host , aws_access_key_id = 'key_id' , aws_secret_access_key = 'secret' ) #NEWLINE# awsc . _mexe ( 'PUT' , path , None , headers , sender = sender ) #NEWLINE# metadata = self . ckan . storage_metadata_update ( path , { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } ) #NEWLINE# from pprint import pprint #NEWLINE# pprint ( metadata )