<s> from __future__ import unicode_literals <EOL> import hashlib <EOL> import time <EOL> import re <EOL> from xml . sax . saxutils import escape <EOL> import boto . sqs <EOL> from moto . core import BaseBackend <EOL> from moto . core . utils import camelcase_to_underscores , get_random_message_id , unix_time_millis <EOL> from . utils import generate_receipt_handle <EOL> from . exceptions import ( <EOL> ReceiptHandleIsInvalid , <EOL> MessageNotInflight <EOL> ) <EOL> DEFAULT_ACCOUNT_ID = <NUM_LIT> <EOL> class Message ( object ) : <EOL> def __init__ ( self , message_id , body ) : <EOL> self . id = message_id <EOL> self . _body = body <EOL> self . message_attributes = { } <EOL> self . receipt_handle = None <EOL> self . sender_id = DEFAULT_ACCOUNT_ID <EOL> self . sent_timestamp = None <EOL> self . approximate_first_receive_timestamp = None <EOL> self . approximate_receive_count = <NUM_LIT:0> <EOL> self . visible_at = <NUM_LIT:0> <EOL> self . delayed_until = <NUM_LIT:0> <EOL> @ property <EOL> def md5 ( self ) : <EOL> body_md5 = hashlib . md5 ( ) <EOL> body_md5 . update ( self . _body . encode ( '<STR_LIT:utf-8>' ) ) <EOL> return body_md5 . hexdigest ( ) <EOL> @ property <EOL> def body ( self ) : <EOL> return escape ( self . _body ) <EOL> def mark_sent ( self , delay_seconds = None ) : <EOL> self . sent_timestamp = unix_time_millis ( ) <EOL> if delay_seconds : <EOL> self . delay ( delay_seconds = delay_seconds ) <EOL> def mark_received ( self , visibility_timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if visibility_timeout : <EOL> visibility_timeout = int ( visibility_timeout ) <EOL> else : <EOL> visibility_timeout = <NUM_LIT:0> <EOL> if not self . approximate_first_receive_timestamp : <EOL> self . approximate_first_receive_timestamp = unix_time_millis ( ) <EOL> self . approximate_receive_count += <NUM_LIT:1> <EOL> if visibility_timeout : <EOL> self . change_visibility ( visibility_timeout ) <EOL> self . receipt_handle = generate_receipt_handle ( ) <EOL> def change_visibility ( self , visibility_timeout ) : <EOL> visibility_timeout_msec = int ( visibility_timeout ) * <NUM_LIT:1000> <EOL> self . visible_at = unix_time_millis ( ) + visibility_timeout_msec <EOL> def delay ( self , delay_seconds ) : <EOL> delay_msec = int ( delay_seconds ) * <NUM_LIT:1000> <EOL> self . delayed_until = unix_time_millis ( ) + delay_msec <EOL> @ property <EOL> def visible ( self ) : <EOL> current_time = unix_time_millis ( ) <EOL> if current_time > self . visible_at : <EOL> return True <EOL> return False <EOL> @ property <EOL> def delayed ( self ) : <EOL> current_time = unix_time_millis ( ) <EOL> if current_time < self . delayed_until : <EOL> return True <EOL> return False <EOL> class Queue ( object ) : <EOL> camelcase_attributes = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> def __init__ ( self , name , visibility_timeout , wait_time_seconds , region ) : <EOL> self . name = name <EOL> self . visibility_timeout = visibility_timeout or <NUM_LIT:30> <EOL> self . region = region <EOL> self . wait_time_seconds = wait_time_seconds or <NUM_LIT:0> <EOL> self . _messages = [ ] <EOL> now = time . time ( ) <EOL> self . created_timestamp = now <EOL> self . delay_seconds = <NUM_LIT:0> <EOL> self . last_modified_timestamp = now <EOL> self . maximum_message_size = <NUM_LIT:64> << <NUM_LIT:10> <EOL> self . message_retention_period = <NUM_LIT> * <NUM_LIT:4> <EOL> self . queue_arn = '<STR_LIT>' % self . name <EOL> self . receive_message_wait_time_seconds = <NUM_LIT:0> <EOL> @ classmethod <EOL> def create_from_cloudformation_json ( cls , resource_name , cloudformation_json , region_name ) : <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> sqs_backend = sqs_backends [ region_name ] <EOL> return sqs_backend . create_queue ( <EOL> name = properties [ '<STR_LIT>' ] , <EOL> visibility_timeout = properties . get ( '<STR_LIT>' ) , <EOL> wait_time_seconds = properties . get ( '<STR_LIT>' ) <EOL> ) <EOL> @ classmethod <EOL> def update_from_cloudformation_json ( cls , original_resource , new_resource_name , cloudformation_json , region_name ) : <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> queue_name = properties [ '<STR_LIT>' ] <EOL> sqs_backend = sqs_backends [ region_name ] <EOL> queue = sqs_backend . get_queue ( queue_name ) <EOL> if '<STR_LIT>' in properties : <EOL> queue . visibility_timeout = int ( properties [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in properties : <EOL> queue . wait_time_seconds = int ( properties [ '<STR_LIT>' ] ) <EOL> return queue <EOL> @ classmethod <EOL> def delete_from_cloudformation_json ( cls , resource_name , cloudformation_json , region_name ) : <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> queue_name = properties [ '<STR_LIT>' ] <EOL> sqs_backend = sqs_backends [ region_name ] <EOL> sqs_backend . delete_queue ( queue_name ) <EOL> @ property <EOL> def approximate_number_of_messages_delayed ( self ) : <EOL> return len ( [ m for m in self . _messages if m . delayed ] ) <EOL> @ property <EOL> def approximate_number_of_messages_not_visible ( self ) : <EOL> return len ( [ m for m in self . _messages if not m . visible ] ) <EOL> @ property <EOL> def approximate_number_of_messages ( self ) : <EOL> return len ( self . messages ) <EOL> @ property <EOL> def physical_resource_id ( self ) : <EOL> return self . name <EOL> @ property <EOL> def attributes ( self ) : <EOL> result = { } <EOL> for attribute in self . camelcase_attributes : <EOL> result [ attribute ] = getattr ( self , camelcase_to_underscores ( attribute ) ) <EOL> return result <EOL> @ property <EOL> def url ( self ) : <EOL> return "<STR_LIT>" . format ( self . region , self . name ) <EOL> @ property <EOL> def messages ( self ) : <EOL> return [ message for message in self . _messages if message . visible and not message . delayed ] <EOL> def add_message ( self , message ) : <EOL> self . _messages . append ( message ) <EOL> def get_cfn_attribute ( self , attribute_name ) : <EOL> from moto . cloudformation . exceptions import UnformattedGetAttTemplateException <EOL> if attribute_name == '<STR_LIT>' : <EOL> return self . queue_arn <EOL> elif attribute_name == '<STR_LIT>' : <EOL> return self . name <EOL> raise UnformattedGetAttTemplateException ( ) <EOL> class SQSBackend ( BaseBackend ) : <EOL> def __init__ ( self , region_name ) : <EOL> self . region_name = region_name <EOL> self . queues = { } <EOL> super ( SQSBackend , self ) . __init__ ( ) <EOL> def reset ( self ) : <EOL> region_name = self . region_name <EOL> self . __dict__ = { } <EOL> self . __init__ ( region_name ) <EOL> def create_queue ( self , name , visibility_timeout , wait_time_seconds ) : <EOL> queue = self . queues . get ( name ) <EOL> if queue is None : <EOL> queue = Queue ( name , visibility_timeout , wait_time_seconds , self . region_name ) <EOL> self . queues [ name ] = queue <EOL> return queue <EOL> def list_queues ( self , queue_name_prefix ) : <EOL> re_str = '<STR_LIT>' <EOL> if queue_name_prefix : <EOL> re_str = '<STR_LIT>' . format ( queue_name_prefix ) <EOL> prefix_re = re . compile ( re_str ) <EOL> qs = [ ] <EOL> for name , q in self . queues . items ( ) : <EOL> if prefix_re . search ( name ) : <EOL> qs . append ( q ) <EOL> return qs <EOL> def get_queue ( self , queue_name ) : <EOL> return self . queues . get ( queue_name , None ) <EOL> def delete_queue ( self , queue_name ) : <EOL> if queue_name in self . queues : <EOL> return self . queues . pop ( queue_name ) <EOL> return False <EOL> def set_queue_attribute ( self , queue_name , key , value ) : <EOL> queue = self . get_queue ( queue_name ) <EOL> setattr ( queue , key , value ) <EOL> return queue <EOL> def send_message ( self , queue_name , message_body , message_attributes = None , delay_seconds = None ) : <EOL> queue = self . get_queue ( queue_name ) <EOL> if delay_seconds : <EOL> delay_seconds = int ( delay_seconds ) <EOL> else : <EOL> delay_seconds = queue . delay_seconds <EOL> message_id = get_random_message_id ( ) <EOL> message = Message ( message_id , message_body ) <EOL> if message_attributes : <EOL> message . message_attributes = message_attributes <EOL> message . mark_sent ( <EOL> delay_seconds = delay_seconds <EOL> ) <EOL> queue . add_message ( message ) <EOL> return message <EOL> def receive_messages ( self , queue_name , count , wait_seconds_timeout ) : <EOL> """<STR_LIT>""" <EOL> queue = self . get_queue ( queue_name ) <EOL> result = [ ] <EOL> polling_end = time . time ( ) + wait_seconds_timeout <EOL> while True : <EOL> for message in queue . messages : <EOL> if not message . visible : <EOL> continue <EOL> message . mark_received ( <EOL> visibility_timeout = queue . visibility_timeout <EOL> ) <EOL> result . append ( message ) <EOL> if len ( result ) >= count : <EOL> break <EOL> if result or time . time ( ) > polling_end : <EOL> break <EOL> return result <EOL> def delete_message ( self , queue_name , receipt_handle ) : <EOL> queue = self . get_queue ( queue_name ) <EOL> new_messages = [ ] <EOL> for message in queue . _messages : <EOL> if message . receipt_handle == receipt_handle : <EOL> continue <EOL> new_messages . append ( message ) <EOL> queue . _messages = new_messages <EOL> def change_message_visibility ( self , queue_name , receipt_handle , visibility_timeout ) : <EOL> queue = self . get_queue ( queue_name ) <EOL> for message in queue . _messages : <EOL> if message . receipt_handle == receipt_handle : <EOL> if message . visible : <EOL> raise MessageNotInflight <EOL> message . change_visibility ( visibility_timeout ) <EOL> return <EOL> raise ReceiptHandleIsInvalid <EOL> def purge_queue ( self , queue_name ) : <EOL> queue = self . get_queue ( queue_name ) <EOL> queue . _messages = [ ] <EOL> sqs_backends = { } <EOL> for region in boto . sqs . regions ( ) : <EOL> sqs_backends [ region . name ] = SQSBackend ( region . name ) </s>
