<s> import ufora . distributed . S3 . S3Interface as S3Interface <EOL> import collections <EOL> import logging <EOL> import threading <EOL> import time <EOL> InMemoryS3Key = collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT:value>' , '<STR_LIT>' ] ) <EOL> class InMemoryS3State ( object ) : <EOL> def __init__ ( self ) : <EOL> self . writeFailureInjector = None <EOL> self . delayAfterWriteFunction = None <EOL> self . throughput = None <EOL> self . clear ( ) <EOL> def clear ( self ) : <EOL> self . buckets_ = { } <EOL> self . bucketOwners_ = { } <EOL> self . lock = threading . RLock ( ) <EOL> self . throughput = None <EOL> self . bytesLoadedPerMachine = { } <EOL> def logBytesLoadedAndDelay ( self , machine , data ) : <EOL> with self . lock : <EOL> if machine not in self . bytesLoadedPerMachine : <EOL> self . bytesLoadedPerMachine [ machine ] = <NUM_LIT:0> <EOL> self . bytesLoadedPerMachine [ machine ] += data <EOL> if self . throughput is not None : <EOL> time . sleep ( float ( data ) / self . throughput ) <EOL> def setThroughputPerMachine ( self , throughput ) : <EOL> self . throughput = throughput <EOL> def createBucket ( self , bucketName , credentials ) : <EOL> assert bucketName not in self . bucketOwners_ <EOL> self . bucketOwners_ [ bucketName ] = credentials <EOL> self . buckets_ [ bucketName ] = { } <EOL> def validateAccess ( self , bucketName , credentials ) : <EOL> owner = self . bucketOwners_ . get ( bucketName ) <EOL> if owner is None : <EOL> raise S3Interface . BucketNotFound ( bucketName ) <EOL> if credentials != owner and owner != S3Interface . S3Interface . publicCredentials : <EOL> logging . error ( "<STR_LIT>" , <EOL> owner , credentials ) <EOL> raise S3Interface . BucketAccessError ( bucketName ) <EOL> def listBuckets ( self , credentials ) : <EOL> return [ <EOL> name for name , owner in self . bucketOwners_ . iteritems ( ) <EOL> if owner == credentials or owner == self . publicCredentials <EOL> ] <EOL> class InMemoryS3InterfaceFactory ( S3Interface . S3InterfaceFactory ) : <EOL> isCompatibleWithOutOfProcessDownloadPool = False <EOL> def __init__ ( self , state = None , onMachine = None ) : <EOL> self . state_ = state or InMemoryS3State ( ) <EOL> self . machine_ = onMachine <EOL> def setWriteFailureInjector ( self , injector ) : <EOL> """<STR_LIT>""" <EOL> self . state_ . writeFailureInjector = injector <EOL> def setDelayAfterWriteInjector ( self , injector ) : <EOL> """<STR_LIT>""" <EOL> self . state_ . delayAfterWriteFunction = injector <EOL> def clearState ( self ) : <EOL> self . state_ . clear ( ) <EOL> def withMachine ( self , machine ) : <EOL> return InMemoryS3InterfaceFactory ( self . state_ , machine ) <EOL> def setThroughputPerMachine ( self , bytesPerSecond ) : <EOL> self . state_ . setThroughputPerMachine ( bytesPerSecond ) <EOL> def getPerMachineBytecounts ( self ) : <EOL> return dict ( self . state_ . bytesLoadedPerMachine ) <EOL> def totalBytesInStorage ( self ) : <EOL> total = <NUM_LIT:0> <EOL> for bucket in self . state_ . buckets_ : <EOL> for key in self . state_ . buckets_ [ bucket ] : <EOL> total += len ( self . state_ . buckets_ [ bucket ] [ key ] . value ) <EOL> return total <EOL> def listAllBucketKeyPairs ( self ) : <EOL> res = [ ] <EOL> for bucket in self . state_ . buckets_ : <EOL> for key in self . state_ . buckets_ [ bucket ] : <EOL> res . append ( ( bucket , key ) ) <EOL> return res <EOL> def __call__ ( self , awsAccessKey = '<STR_LIT>' , awsSecretKey = '<STR_LIT>' ) : <EOL> return InMemoryS3Interface ( self . state_ , self . machine_ , awsAccessKey , awsSecretKey ) <EOL> class InMemoryS3Interface ( S3Interface . S3Interface ) : <EOL> def __init__ ( self , state , machine , awsAccessKey = '<STR_LIT>' , awsSecretKey = '<STR_LIT>' ) : <EOL> S3Interface . S3Interface . __init__ ( self ) <EOL> self . credentials_ = ( awsAccessKey , awsSecretKey ) <EOL> self . state_ = state <EOL> self . machine_ = machine <EOL> def initiateMultipartUpload ( self , bucketName , eventualKeyName ) : <EOL> prefix = "<STR_LIT>" + eventualKeyName <EOL> return prefix <EOL> def completeMultipartUpload ( self , bucketName , eventualKeyName , uploadId ) : <EOL> keys = self . listKeysWithPrefix ( bucketName , uploadId ) <EOL> partIdsAndKeys = sorted ( [ ( int ( keyname [ len ( uploadId ) : ] ) , keyname ) for keyname , _ , _ in keys ] ) <EOL> res = "<STR_LIT>" <EOL> assert len ( partIdsAndKeys ) <= <NUM_LIT:1000> <EOL> for _ , keyname in partIdsAndKeys : <EOL> res = res + self . getKeyValue ( bucketName , keyname ) <EOL> self . deleteKey ( bucketName , keyname ) <EOL> self . setKeyValue ( bucketName , eventualKeyName , res ) <EOL> def setMultipartUploadPart ( self , bucketName , eventualKeyName , uploadId , oneBasedPartNumber , value ) : <EOL> """<STR_LIT>""" <EOL> assert oneBasedPartNumber >= <NUM_LIT:1> and oneBasedPartNumber <= <NUM_LIT> <EOL> self . setKeyValue ( bucketName , uploadId + str ( oneBasedPartNumber ) , value ) <EOL> def listBuckets ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . state_ . lock : <EOL> return self . state_ . listBuckets ( self . credentials_ ) <EOL> def listKeysAndSizes ( self , bucketName ) : <EOL> """<STR_LIT>""" <EOL> with self . state_ . lock : <EOL> if bucketName not in self . state_ . buckets_ : <EOL> raise S3Interface . BucketNotFound ( bucketName ) <EOL> self . state_ . validateAccess ( bucketName , self . credentials_ ) <EOL> return [ ( key , len ( val . value ) , val . mtime ) <EOL> for key , val in self . state_ . buckets_ [ bucketName ] . iteritems ( ) ] <EOL> def listKeysWithPrefix ( self , bucketName , prefix ) : <EOL> """<STR_LIT>""" <EOL> with self . state_ . lock : <EOL> self . state_ . validateAccess ( bucketName , self . credentials_ ) <EOL> return [ ( key , sz , mtime ) <EOL> for key , sz , mtime in self . listKeysAndSizes ( bucketName ) <EOL> if key . startswith ( prefix ) <EOL> ] <EOL> def getKeyValue ( self , bucketName , keyName , logBytes = True ) : <EOL> """<STR_LIT>""" <EOL> with self . state_ . lock : <EOL> if bucketName not in self . state_ . buckets_ : <EOL> raise S3Interface . BucketNotFound ( bucketName ) <EOL> self . state_ . validateAccess ( bucketName , self . credentials_ ) <EOL> if keyName not in self . state_ . buckets_ [ bucketName ] : <EOL> raise S3Interface . KeyNotFound ( bucketName , keyName ) <EOL> result = self . state_ . buckets_ [ bucketName ] [ keyName ] . value <EOL> if logBytes : <EOL> self . state_ . logBytesLoadedAndDelay ( self . machine_ , len ( result ) ) <EOL> return result <EOL> def deleteKey ( self , bucketName , keyName ) : <EOL> """<STR_LIT>""" <EOL> with self . state_ . lock : <EOL> if bucketName not in self . state_ . buckets_ : <EOL> self . state_ . createBucket ( bucketName , self . credentials_ ) <EOL> self . state_ . validateAccess ( bucketName , self . credentials_ ) <EOL> if keyName not in self . state_ . buckets_ [ bucketName ] : <EOL> raise S3Interface . KeyNotFound ( bucketName , keyName ) <EOL> del self . state_ . buckets_ [ bucketName ] [ keyName ] <EOL> def getKeyValueOverRange ( self , bucketName , keyName , lowIndex , highIndex ) : <EOL> """<STR_LIT>""" <EOL> with self . state_ . lock : <EOL> self . state_ . validateAccess ( bucketName , self . credentials_ ) <EOL> result = self . getKeyValue ( bucketName , keyName , logBytes = False ) [ lowIndex : highIndex ] <EOL> self . state_ . logBytesLoadedAndDelay ( self . machine_ , len ( result ) ) <EOL> return result <EOL> def getKeySize ( self , bucketName , keyName ) : <EOL> """<STR_LIT>""" <EOL> with self . state_ . lock : <EOL> self . state_ . validateAccess ( bucketName , self . credentials_ ) <EOL> return len ( self . getKeyValue ( bucketName , keyName , logBytes = False ) ) <EOL> def keyExists ( self , bucketName , keyName ) : <EOL> """<STR_LIT>""" <EOL> with self . state_ . lock : <EOL> self . state_ . validateAccess ( bucketName , self . credentials_ ) <EOL> if bucketName not in self . state_ . buckets_ : <EOL> return False <EOL> return keyName in self . state_ . buckets_ [ bucketName ] <EOL> def bucketExists ( self , bucketName ) : <EOL> """<STR_LIT>""" <EOL> with self . state_ . lock : <EOL> return bucketName in self . state_ . buckets_ <EOL> def setKeyValue ( self , bucketName , keyName , value ) : <EOL> """<STR_LIT>""" <EOL> with self . state_ . lock : <EOL> if self . state_ . writeFailureInjector is not None : <EOL> if self . state_ . writeFailureInjector ( bucketName , keyName ) : <EOL> raise S3Interface . UnableToWriteKey ( bucketName , keyName ) <EOL> assert isinstance ( value , str ) <EOL> if bucketName not in self . state_ . buckets_ : <EOL> self . state_ . createBucket ( bucketName , self . credentials_ ) <EOL> self . state_ . validateAccess ( bucketName , self . credentials_ ) <EOL> self . state_ . buckets_ [ bucketName ] [ keyName ] = InMemoryS3Key ( value = value , <EOL> mtime = time . time ( ) ) <EOL> if self . state_ . delayAfterWriteFunction is not None : <EOL> delay = self . state_ . delayAfterWriteFunction ( bucketName , keyName ) <EOL> time . sleep ( delay ) <EOL> def setKeyValueFromFile ( self , bucketName , keyName , filePath ) : <EOL> """<STR_LIT>""" <EOL> pass </s>
