<s> from __future__ import unicode_literals <EOL> import datetime <EOL> import time <EOL> import boto . kinesis <EOL> import re <EOL> import six <EOL> import itertools <EOL> from operator import attrgetter <EOL> from hashlib import md5 <EOL> from moto . compat import OrderedDict <EOL> from moto . core import BaseBackend <EOL> from . exceptions import StreamNotFoundError , ShardNotFoundError , ResourceInUseError , ResourceNotFoundError , InvalidArgumentError <EOL> from . utils import compose_shard_iterator , compose_new_shard_iterator , decompose_shard_iterator <EOL> class Record ( object ) : <EOL> def __init__ ( self , partition_key , data , sequence_number , explicit_hash_key ) : <EOL> self . partition_key = partition_key <EOL> self . data = data <EOL> self . sequence_number = sequence_number <EOL> self . explicit_hash_key = explicit_hash_key <EOL> def to_json ( self ) : <EOL> return { <EOL> "<STR_LIT>" : self . data , <EOL> "<STR_LIT>" : self . partition_key , <EOL> "<STR_LIT>" : str ( self . sequence_number ) , <EOL> } <EOL> class Shard ( object ) : <EOL> def __init__ ( self , shard_id , starting_hash , ending_hash ) : <EOL> self . _shard_id = shard_id <EOL> self . starting_hash = starting_hash <EOL> self . ending_hash = ending_hash <EOL> self . records = OrderedDict ( ) <EOL> @ property <EOL> def shard_id ( self ) : <EOL> return "<STR_LIT>" . format ( str ( self . _shard_id ) . zfill ( <NUM_LIT:12> ) ) <EOL> def get_records ( self , last_sequence_id , limit ) : <EOL> last_sequence_id = int ( last_sequence_id ) <EOL> results = [ ] <EOL> for sequence_number , record in self . records . items ( ) : <EOL> if sequence_number > last_sequence_id : <EOL> results . append ( record ) <EOL> last_sequence_id = sequence_number <EOL> if len ( results ) == limit : <EOL> break <EOL> return results , last_sequence_id <EOL> def put_record ( self , partition_key , data , explicit_hash_key ) : <EOL> if self . records : <EOL> last_sequence_number = self . get_max_sequence_number ( ) <EOL> else : <EOL> last_sequence_number = <NUM_LIT:0> <EOL> sequence_number = last_sequence_number + <NUM_LIT:1> <EOL> self . records [ sequence_number ] = Record ( partition_key , data , sequence_number , explicit_hash_key ) <EOL> return sequence_number <EOL> def get_min_sequence_number ( self ) : <EOL> if self . records : <EOL> return list ( self . records . keys ( ) ) [ <NUM_LIT:0> ] <EOL> return <NUM_LIT:0> <EOL> def get_max_sequence_number ( self ) : <EOL> if self . records : <EOL> return list ( self . records . keys ( ) ) [ - <NUM_LIT:1> ] <EOL> return <NUM_LIT:0> <EOL> def to_json ( self ) : <EOL> return { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : str ( self . ending_hash ) , <EOL> "<STR_LIT>" : str ( self . starting_hash ) <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . get_max_sequence_number ( ) , <EOL> "<STR_LIT>" : self . get_min_sequence_number ( ) , <EOL> } , <EOL> "<STR_LIT>" : self . shard_id <EOL> } <EOL> class Stream ( object ) : <EOL> def __init__ ( self , stream_name , shard_count , region ) : <EOL> self . stream_name = stream_name <EOL> self . shard_count = shard_count <EOL> self . region = region <EOL> self . account_number = "<STR_LIT>" <EOL> self . shards = { } <EOL> self . tags = { } <EOL> if six . PY3 : <EOL> izip_longest = itertools . zip_longest <EOL> else : <EOL> izip_longest = itertools . izip_longest <EOL> for index , start , end in izip_longest ( range ( shard_count ) , <EOL> range ( <NUM_LIT:0> , <NUM_LIT:2> ** <NUM_LIT> , <NUM_LIT:2> ** <NUM_LIT> // shard_count ) , <EOL> range ( <NUM_LIT:2> ** <NUM_LIT> // shard_count , <NUM_LIT:2> ** <NUM_LIT> , <NUM_LIT:2> ** <NUM_LIT> // shard_count ) , <EOL> fillvalue = <NUM_LIT:2> ** <NUM_LIT> ) : <EOL> shard = Shard ( index , start , end ) <EOL> self . shards [ shard . shard_id ] = shard <EOL> @ property <EOL> def arn ( self ) : <EOL> return "<STR_LIT>" . format ( <EOL> region = self . region , <EOL> account_number = self . account_number , <EOL> stream_name = self . stream_name <EOL> ) <EOL> def get_shard ( self , shard_id ) : <EOL> if shard_id in self . shards : <EOL> return self . shards [ shard_id ] <EOL> else : <EOL> raise ShardNotFoundError ( shard_id ) <EOL> def get_shard_for_key ( self , partition_key , explicit_hash_key ) : <EOL> if not isinstance ( partition_key , six . string_types ) : <EOL> raise InvalidArgumentError ( "<STR_LIT>" ) <EOL> if len ( partition_key ) > <NUM_LIT> : <EOL> raise InvalidArgumentError ( "<STR_LIT>" ) <EOL> if explicit_hash_key : <EOL> if not isinstance ( explicit_hash_key , six . string_types ) : <EOL> raise InvalidArgumentError ( "<STR_LIT>" ) <EOL> key = int ( explicit_hash_key ) <EOL> if key >= <NUM_LIT:2> ** <NUM_LIT> : <EOL> raise InvalidArgumentError ( "<STR_LIT>" ) <EOL> else : <EOL> key = int ( md5 ( partition_key . encode ( '<STR_LIT:utf-8>' ) ) . hexdigest ( ) , <NUM_LIT:16> ) <EOL> for shard in self . shards . values ( ) : <EOL> if shard . starting_hash <= key < shard . ending_hash : <EOL> return shard <EOL> def put_record ( self , partition_key , explicit_hash_key , sequence_number_for_ordering , data ) : <EOL> shard = self . get_shard_for_key ( partition_key , explicit_hash_key ) <EOL> sequence_number = shard . put_record ( partition_key , data , explicit_hash_key ) <EOL> return sequence_number , shard . shard_id <EOL> def to_json ( self ) : <EOL> return { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . arn , <EOL> "<STR_LIT>" : self . stream_name , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : [ shard . to_json ( ) for shard in self . shards . values ( ) ] , <EOL> } <EOL> } <EOL> class FirehoseRecord ( object ) : <EOL> def __init__ ( self , record_data ) : <EOL> self . record_id = <NUM_LIT> <EOL> self . record_data = record_data <EOL> class DeliveryStream ( object ) : <EOL> def __init__ ( self , stream_name , ** stream_kwargs ) : <EOL> self . name = stream_name <EOL> self . redshift_username = stream_kwargs [ '<STR_LIT>' ] <EOL> self . redshift_password = stream_kwargs [ '<STR_LIT>' ] <EOL> self . redshift_jdbc_url = stream_kwargs [ '<STR_LIT>' ] <EOL> self . redshift_role_arn = stream_kwargs [ '<STR_LIT>' ] <EOL> self . redshift_copy_command = stream_kwargs [ '<STR_LIT>' ] <EOL> self . redshift_s3_role_arn = stream_kwargs [ '<STR_LIT>' ] <EOL> self . redshift_s3_bucket_arn = stream_kwargs [ '<STR_LIT>' ] <EOL> self . redshift_s3_prefix = stream_kwargs [ '<STR_LIT>' ] <EOL> self . redshift_s3_compression_format = stream_kwargs . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . redshift_s3_buffering_hings = stream_kwargs [ '<STR_LIT>' ] <EOL> self . records = [ ] <EOL> self . status = '<STR_LIT>' <EOL> self . create_at = datetime . datetime . utcnow ( ) <EOL> self . last_updated = datetime . datetime . utcnow ( ) <EOL> @ property <EOL> def arn ( self ) : <EOL> return '<STR_LIT>' . format ( self . name ) <EOL> def to_dict ( self ) : <EOL> return { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : time . mktime ( self . create_at . timetuple ( ) ) , <EOL> "<STR_LIT>" : self . arn , <EOL> "<STR_LIT>" : self . name , <EOL> "<STR_LIT>" : self . status , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT>" : "<STR_LIT:string>" , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . redshift_jdbc_url , <EOL> "<STR_LIT>" : self . redshift_copy_command , <EOL> "<STR_LIT>" : self . redshift_role_arn , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . redshift_s3_bucket_arn , <EOL> "<STR_LIT>" : self . redshift_s3_buffering_hings , <EOL> "<STR_LIT>" : self . redshift_s3_compression_format , <EOL> "<STR_LIT>" : self . redshift_s3_prefix , <EOL> "<STR_LIT>" : self . redshift_s3_role_arn <EOL> } , <EOL> "<STR_LIT>" : self . redshift_username , <EOL> } , <EOL> } <EOL> ] , <EOL> "<STR_LIT>" : False , <EOL> "<STR_LIT>" : time . mktime ( self . last_updated . timetuple ( ) ) , <EOL> "<STR_LIT>" : "<STR_LIT:string>" , <EOL> } <EOL> } <EOL> def put_record ( self , record_data ) : <EOL> record = FirehoseRecord ( record_data ) <EOL> self . records . append ( record ) <EOL> return record <EOL> class KinesisBackend ( BaseBackend ) : <EOL> def __init__ ( self ) : <EOL> self . streams = { } <EOL> self . delivery_streams = { } <EOL> def create_stream ( self , stream_name , shard_count , region ) : <EOL> if stream_name in self . streams : <EOL> raise ResourceInUseError ( stream_name ) <EOL> stream = Stream ( stream_name , shard_count , region ) <EOL> self . streams [ stream_name ] = stream <EOL> return stream <EOL> def describe_stream ( self , stream_name ) : <EOL> if stream_name in self . streams : <EOL> return self . streams [ stream_name ] <EOL> else : <EOL> raise StreamNotFoundError ( stream_name ) <EOL> def list_streams ( self ) : <EOL> return self . streams . values ( ) <EOL> def delete_stream ( self , stream_name ) : <EOL> if stream_name in self . streams : <EOL> return self . streams . pop ( stream_name ) <EOL> raise StreamNotFoundError ( stream_name ) <EOL> def get_shard_iterator ( self , stream_name , shard_id , shard_iterator_type , starting_sequence_number ) : <EOL> stream = self . describe_stream ( stream_name ) <EOL> shard = stream . get_shard ( shard_id ) <EOL> shard_iterator = compose_new_shard_iterator ( <EOL> stream_name , shard , shard_iterator_type , starting_sequence_number <EOL> ) <EOL> return shard_iterator <EOL> def get_records ( self , shard_iterator , limit ) : <EOL> decomposed = decompose_shard_iterator ( shard_iterator ) <EOL> stream_name , shard_id , last_sequence_id = decomposed <EOL> stream = self . describe_stream ( stream_name ) <EOL> shard = stream . get_shard ( shard_id ) <EOL> records , last_sequence_id = shard . get_records ( last_sequence_id , limit ) <EOL> next_shard_iterator = compose_shard_iterator ( stream_name , shard , last_sequence_id ) <EOL> return next_shard_iterator , records <EOL> def put_record ( self , stream_name , partition_key , explicit_hash_key , sequence_number_for_ordering , data ) : <EOL> stream = self . describe_stream ( stream_name ) <EOL> sequence_number , shard_id = stream . put_record ( <EOL> partition_key , explicit_hash_key , sequence_number_for_ordering , data <EOL> ) <EOL> return sequence_number , shard_id <EOL> def put_records ( self , stream_name , records ) : <EOL> stream = self . describe_stream ( stream_name ) <EOL> response = { <EOL> "<STR_LIT>" : <NUM_LIT:0> , <EOL> "<STR_LIT>" : [ ] <EOL> } <EOL> for record in records : <EOL> partition_key = record . get ( "<STR_LIT>" ) <EOL> explicit_hash_key = record . get ( "<STR_LIT>" ) <EOL> data = record . get ( "<STR_LIT>" ) <EOL> sequence_number , shard_id = stream . put_record ( <EOL> partition_key , explicit_hash_key , None , data <EOL> ) <EOL> response [ '<STR_LIT>' ] . append ( { <EOL> "<STR_LIT>" : sequence_number , <EOL> "<STR_LIT>" : shard_id <EOL> } ) <EOL> return response <EOL> def split_shard ( self , stream_name , shard_to_split , new_starting_hash_key ) : <EOL> stream = self . describe_stream ( stream_name ) <EOL> if shard_to_split not in stream . shards : <EOL> raise ResourceNotFoundError ( shard_to_split ) <EOL> if not re . match ( r'<STR_LIT>' , new_starting_hash_key ) : <EOL> raise InvalidArgumentError ( new_starting_hash_key ) <EOL> new_starting_hash_key = int ( new_starting_hash_key ) <EOL> shard = stream . shards [ shard_to_split ] <EOL> last_id = sorted ( stream . shards . values ( ) , key = attrgetter ( '<STR_LIT>' ) ) [ - <NUM_LIT:1> ] . _shard_id <EOL> if shard . starting_hash < new_starting_hash_key < shard . ending_hash : <EOL> new_shard = Shard ( last_id + <NUM_LIT:1> , new_starting_hash_key , shard . ending_hash ) <EOL> shard . ending_hash = new_starting_hash_key <EOL> stream . shards [ new_shard . shard_id ] = new_shard <EOL> else : <EOL> raise InvalidArgumentError ( new_starting_hash_key ) <EOL> records = shard . records <EOL> shard . records = OrderedDict ( ) <EOL> for index in records : <EOL> record = records [ index ] <EOL> stream . put_record ( <EOL> record . partition_key , record . explicit_hash_key , None , record . data <EOL> ) <EOL> def merge_shards ( self , stream_name , shard_to_merge , adjacent_shard_to_merge ) : <EOL> stream = self . describe_stream ( stream_name ) <EOL> if shard_to_merge not in stream . shards : <EOL> raise ResourceNotFoundError ( shard_to_merge ) <EOL> if adjacent_shard_to_merge not in stream . shards : <EOL> raise ResourceNotFoundError ( adjacent_shard_to_merge ) <EOL> shard1 = stream . shards [ shard_to_merge ] <EOL> shard2 = stream . shards [ adjacent_shard_to_merge ] <EOL> if shard1 . ending_hash == shard2 . starting_hash : <EOL> shard1 . ending_hash = shard2 . ending_hash <EOL> elif shard2 . ending_hash == shard1 . starting_hash : <EOL> shard1 . starting_hash = shard2 . starting_hash <EOL> else : <EOL> raise InvalidArgumentError ( adjacent_shard_to_merge ) <EOL> del stream . shards [ shard2 . shard_id ] <EOL> for index in shard2 . records : <EOL> record = shard2 . records [ index ] <EOL> shard1 . put_record ( record . partition_key , record . data , record . explicit_hash_key ) <EOL> '''<STR_LIT>''' <EOL> def create_delivery_stream ( self , stream_name , ** stream_kwargs ) : <EOL> stream = DeliveryStream ( stream_name , ** stream_kwargs ) <EOL> self . delivery_streams [ stream_name ] = stream <EOL> return stream <EOL> def get_delivery_stream ( self , stream_name ) : <EOL> if stream_name in self . delivery_streams : <EOL> return self . delivery_streams [ stream_name ] <EOL> else : <EOL> raise StreamNotFoundError ( stream_name ) <EOL> def list_delivery_streams ( self ) : <EOL> return self . delivery_streams . values ( ) <EOL> def delete_delivery_stream ( self , stream_name ) : <EOL> self . delivery_streams . pop ( stream_name ) <EOL> def put_firehose_record ( self , stream_name , record_data ) : <EOL> stream = self . get_delivery_stream ( stream_name ) <EOL> record = stream . put_record ( record_data ) <EOL> return record <EOL> def list_tags_for_stream ( self , stream_name , exclusive_start_tag_key = None , limit = None ) : <EOL> stream = self . describe_stream ( stream_name ) <EOL> tags = [ ] <EOL> result = { <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : tags <EOL> } <EOL> for key , val in sorted ( stream . tags . items ( ) , key = lambda x : x [ <NUM_LIT:0> ] ) : <EOL> if limit and len ( res ) >= limit : <EOL> result [ '<STR_LIT>' ] = True <EOL> break <EOL> if exclusive_start_tag_key and key < exexclusive_start_tag_key : <EOL> continue <EOL> tags . append ( { <EOL> '<STR_LIT>' : key , <EOL> '<STR_LIT>' : val <EOL> } ) <EOL> return result <EOL> def add_tags_to_stream ( self , stream_name , tags ) : <EOL> stream = self . describe_stream ( stream_name ) <EOL> stream . tags . update ( tags ) <EOL> def remove_tags_from_stream ( self , stream_name , tag_keys ) : <EOL> stream = self . describe_stream ( stream_name ) <EOL> for key in tag_keys : <EOL> if key in stream . tags : <EOL> del stream . tags [ key ] <EOL> kinesis_backends = { } <EOL> for region in boto . kinesis . regions ( ) : <EOL> kinesis_backends [ region . name ] = KinesisBackend ( ) </s>
