Sono disponibili altri esempi AWS SDK nel repository AWS Doc SDK
Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.
Esempi per Amazon SQS con SDK per Python (Boto3)
I seguenti esempi di codice mostrano come eseguire azioni e implementare scenari comuni utilizzando AWS SDK per Python (Boto3) con Amazon SQS.
Le azioni sono estratti di codice da programmi più grandi e devono essere eseguite nel contesto. Sebbene le azioni mostrino come richiamare le singole funzioni del servizio, è possibile visualizzarle contestualizzate negli scenari correlati.
Scenari: esempi di codice che mostrano come eseguire un’attività specifica chiamando più funzioni all’interno dello stesso servizio o combinate con altri Servizi AWS.
Ogni esempio include un link al codice sorgente completo, in cui vengono fornite le istruzioni su come configurare ed eseguire il codice nel contesto.
Argomenti
Azioni
Il seguente esempio di codice mostra come usareCreateQueue.
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. def create_queue(name, attributes=None): """ Creates an Amazon SQS queue. :param name: The name of the queue. This is part of the URL assigned to the queue. :param attributes: The attributes of the queue, such as maximum message size or whether it's a FIFO queue. :return: A Queue object that contains metadata about the queue and that can be used to perform queue operations like sending and receiving messages. """ if not attributes: attributes = {} try: queue = sqs.create_queue(QueueName=name, Attributes=attributes) logger.info("Created queue '%s' with URL=%s", name, queue.url) except ClientError as error: logger.exception("Couldn't create queue named '%s'.", name) raise error else: return queueclass SqsWrapper: """Wrapper class for managing Amazon SQS operations.""" def __init__(self, sqs_client: Any) -> None: """ Initialize the SqsWrapper. :param sqs_client: A Boto3 Amazon SQS client. """ self.sqs_client = sqs_client @classmethod def from_client(cls) -> 'SqsWrapper': """ Create an SqsWrapper instance using a default boto3 client. :return: An instance of this class. """ sqs_client = boto3.client('sqs') return cls(sqs_client) def create_queue(self, queue_name: str, is_fifo: bool = False) -> str: """ Create an SQS queue. :param queue_name: The name of the queue to create. :param is_fifo: Whether to create a FIFO queue. :return: The URL of the created queue. :raises ClientError: If the queue creation fails. """ try: # Add .fifo suffix for FIFO queues if is_fifo and not queue_name.endswith('.fifo'): queue_name += '.fifo' attributes = {} if is_fifo: attributes['FifoQueue'] = 'true' response = self.sqs_client.create_queue( QueueName=queue_name, Attributes=attributes ) queue_url = response['QueueUrl'] logger.info(f"Created queue: {queue_name} with URL: {queue_url}") return queue_url except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error creating queue {queue_name}: {error_code} - {e}") raise-
Per i dettagli sull'API, consulta CreateQueue AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. DeleteMessage
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. def delete_message(message): """ Delete a message from a queue. Clients must delete messages after they are received and processed to remove them from the queue. :param message: The message to delete. The message's queue URL is contained in the message's metadata. :return: None """ try: message.delete() logger.info("Deleted message: %s", message.message_id) except ClientError as error: logger.exception("Couldn't delete message: %s", message.message_id) raise error-
Per i dettagli sull'API, consulta DeleteMessage AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. DeleteMessageBatch
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. def delete_messages(queue, messages): """ Delete a batch of messages from a queue in a single request. :param queue: The queue from which to delete the messages. :param messages: The list of messages to delete. :return: The response from SQS that contains the list of successful and failed message deletions. """ try: entries = [ {"Id": str(ind), "ReceiptHandle": msg.receipt_handle} for ind, msg in enumerate(messages) ] response = queue.delete_messages(Entries=entries) if "Successful" in response: for msg_meta in response["Successful"]: logger.info("Deleted %s", messages[int(msg_meta["Id"])].receipt_handle) if "Failed" in response: for msg_meta in response["Failed"]: logger.warning( "Could not delete %s", messages[int(msg_meta["Id"])].receipt_handle ) except ClientError: logger.exception("Couldn't delete messages from queue %s", queue) else: return responseclass SqsWrapper: """Wrapper class for managing Amazon SQS operations.""" def __init__(self, sqs_client: Any) -> None: """ Initialize the SqsWrapper. :param sqs_client: A Boto3 Amazon SQS client. """ self.sqs_client = sqs_client @classmethod def from_client(cls) -> 'SqsWrapper': """ Create an SqsWrapper instance using a default boto3 client. :return: An instance of this class. """ sqs_client = boto3.client('sqs') return cls(sqs_client) def delete_messages(self, queue_url: str, messages: List[Dict[str, Any]]) -> bool: """ Delete messages from an SQS queue in batches. :param queue_url: The URL of the queue. :param messages: List of messages to delete. :return: True if successful. :raises ClientError: If deleting messages fails. """ try: if not messages: return True # Build delete entries for batch delete delete_entries = [] for i, message in enumerate(messages): delete_entries.append({ 'Id': str(i), 'ReceiptHandle': message['ReceiptHandle'] }) # Delete messages in batches of 10 (SQS limit) batch_size = 10 for i in range(0, len(delete_entries), batch_size): batch = delete_entries[i:i + batch_size] response = self.sqs_client.delete_message_batch( QueueUrl=queue_url, Entries=batch ) # Check for failures if 'Failed' in response and response['Failed']: for failed in response['Failed']: logger.warning(f"Failed to delete message: {failed}") logger.info(f"Deleted {len(messages)} messages from {queue_url}") return True except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error deleting messages: {error_code} - {e}") raise-
Per i dettagli sull'API, consulta DeleteMessageBatch AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. DeleteQueue
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. def remove_queue(queue): """ Removes an SQS queue. When run against an AWS account, it can take up to 60 seconds before the queue is actually deleted. :param queue: The queue to delete. :return: None """ try: queue.delete() logger.info("Deleted queue with URL=%s.", queue.url) except ClientError as error: logger.exception("Couldn't delete queue with URL=%s!", queue.url) raise errorclass SqsWrapper: """Wrapper class for managing Amazon SQS operations.""" def __init__(self, sqs_client: Any) -> None: """ Initialize the SqsWrapper. :param sqs_client: A Boto3 Amazon SQS client. """ self.sqs_client = sqs_client @classmethod def from_client(cls) -> 'SqsWrapper': """ Create an SqsWrapper instance using a default boto3 client. :return: An instance of this class. """ sqs_client = boto3.client('sqs') return cls(sqs_client) def delete_queue(self, queue_url: str) -> bool: """ Delete an SQS queue. :param queue_url: The URL of the queue to delete. :return: True if successful. :raises ClientError: If the queue deletion fails. """ try: self.sqs_client.delete_queue(QueueUrl=queue_url) logger.info(f"Deleted queue: {queue_url}") return True except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') if error_code == 'AWS.SimpleQueueService.NonExistentQueue': logger.warning(f"Queue not found: {queue_url}") return True # Already deleted else: logger.error(f"Error deleting queue: {error_code} - {e}") raise-
Per i dettagli sull'API, consulta DeleteQueue AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. GetQueueAttributes
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. class SqsWrapper: """Wrapper class for managing Amazon SQS operations.""" def __init__(self, sqs_client: Any) -> None: """ Initialize the SqsWrapper. :param sqs_client: A Boto3 Amazon SQS client. """ self.sqs_client = sqs_client @classmethod def from_client(cls) -> 'SqsWrapper': """ Create an SqsWrapper instance using a default boto3 client. :return: An instance of this class. """ sqs_client = boto3.client('sqs') return cls(sqs_client) def get_queue_arn(self, queue_url: str) -> str: """ Get the ARN of an SQS queue. :param queue_url: The URL of the queue. :return: The ARN of the queue. :raises ClientError: If getting queue attributes fails. """ try: response = self.sqs_client.get_queue_attributes( QueueUrl=queue_url, AttributeNames=['QueueArn'] ) queue_arn = response['Attributes']['QueueArn'] logger.info(f"Queue ARN for {queue_url}: {queue_arn}") return queue_arn except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error getting queue ARN: {error_code} - {e}") raise-
Per i dettagli sull'API, consulta GetQueueAttributes AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. GetQueueUrl
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. def get_queue(name): """ Gets an SQS queue by name. :param name: The name that was used to create the queue. :return: A Queue object. """ try: queue = sqs.get_queue_by_name(QueueName=name) logger.info("Got queue '%s' with URL=%s", name, queue.url) except ClientError as error: logger.exception("Couldn't get queue named %s.", name) raise error else: return queue-
Per i dettagli sull'API, consulta GetQueueUrl AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. ListQueues
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. def get_queues(prefix=None): """ Gets a list of SQS queues. When a prefix is specified, only queues with names that start with the prefix are returned. :param prefix: The prefix used to restrict the list of returned queues. :return: A list of Queue objects. """ if prefix: queue_iter = sqs.queues.filter(QueueNamePrefix=prefix) else: queue_iter = sqs.queues.all() queues = list(queue_iter) if queues: logger.info("Got queues: %s", ", ".join([q.url for q in queues])) else: logger.warning("No queues found.") return queues-
Per i dettagli sull'API, consulta ListQueues AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. ReceiveMessage
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. def receive_messages(queue, max_number, wait_time): """ Receive a batch of messages in a single request from an SQS queue. :param queue: The queue from which to receive messages. :param max_number: The maximum number of messages to receive. The actual number of messages received might be less. :param wait_time: The maximum time to wait (in seconds) before returning. When this number is greater than zero, long polling is used. This can result in reduced costs and fewer false empty responses. :return: The list of Message objects received. These each contain the body of the message and metadata and custom attributes. """ try: messages = queue.receive_messages( MessageAttributeNames=["All"], MaxNumberOfMessages=max_number, WaitTimeSeconds=wait_time, ) for msg in messages: logger.info("Received message: %s: %s", msg.message_id, msg.body) except ClientError as error: logger.exception("Couldn't receive messages from queue: %s", queue) raise error else: return messagesclass SqsWrapper: """Wrapper class for managing Amazon SQS operations.""" def __init__(self, sqs_client: Any) -> None: """ Initialize the SqsWrapper. :param sqs_client: A Boto3 Amazon SQS client. """ self.sqs_client = sqs_client @classmethod def from_client(cls) -> 'SqsWrapper': """ Create an SqsWrapper instance using a default boto3 client. :return: An instance of this class. """ sqs_client = boto3.client('sqs') return cls(sqs_client) def receive_messages(self, queue_url: str, max_messages: int = 10) -> List[Dict[str, Any]]: """ Receive messages from an SQS queue. :param queue_url: The URL of the queue to receive messages from. :param max_messages: Maximum number of messages to receive (1-10). :return: List of received messages. :raises ClientError: If receiving messages fails. """ try: # Ensure max_messages is within valid range max_messages = max(1, min(10, max_messages)) response = self.sqs_client.receive_message( QueueUrl=queue_url, MaxNumberOfMessages=max_messages, WaitTimeSeconds=2, # Short polling MessageAttributeNames=['All'] ) messages = response.get('Messages', []) logger.info(f"Received {len(messages)} messages from {queue_url}") return messages except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error receiving messages: {error_code} - {e}") raise-
Per i dettagli sull'API, consulta ReceiveMessage AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. SendMessage
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. def send_message(queue, message_body, message_attributes=None): """ Send a message to an Amazon SQS queue. :param queue: The queue that receives the message. :param message_body: The body text of the message. :param message_attributes: Custom attributes of the message. These are key-value pairs that can be whatever you want. :return: The response from SQS that contains the assigned message ID. """ if not message_attributes: message_attributes = {} try: response = queue.send_message( MessageBody=message_body, MessageAttributes=message_attributes ) except ClientError as error: logger.exception("Send message failed: %s", message_body) raise error else: return response-
Per i dettagli sull'API, consulta SendMessage AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. SendMessageBatch
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. def send_messages(queue, messages): """ Send a batch of messages in a single request to an SQS queue. This request may return overall success even when some messages were not sent. The caller must inspect the Successful and Failed lists in the response and resend any failed messages. :param queue: The queue to receive the messages. :param messages: The messages to send to the queue. These are simplified to contain only the message body and attributes. :return: The response from SQS that contains the list of successful and failed messages. """ try: entries = [ { "Id": str(ind), "MessageBody": msg["body"], "MessageAttributes": msg["attributes"], } for ind, msg in enumerate(messages) ] response = queue.send_messages(Entries=entries) if "Successful" in response: for msg_meta in response["Successful"]: logger.info( "Message sent: %s: %s", msg_meta["MessageId"], messages[int(msg_meta["Id"])]["body"], ) if "Failed" in response: for msg_meta in response["Failed"]: logger.warning( "Failed to send: %s: %s", msg_meta["MessageId"], messages[int(msg_meta["Id"])]["body"], ) except ClientError as error: logger.exception("Send messages failed to queue: %s", queue) raise error else: return response-
Per i dettagli sull'API, consulta SendMessageBatch AWSSDK for Python (Boto3) API Reference.
-
Il seguente esempio di codice mostra come utilizzare. SetQueueAttributes
- SDK per Python (Boto3)
-
Nota
C'è altro da fare GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. Imposta l’attributo della policy di una coda per un argomento.
class SqsWrapper: """Wrapper class for managing Amazon SQS operations.""" def __init__(self, sqs_client: Any) -> None: """ Initialize the SqsWrapper. :param sqs_client: A Boto3 Amazon SQS client. """ self.sqs_client = sqs_client @classmethod def from_client(cls) -> 'SqsWrapper': """ Create an SqsWrapper instance using a default boto3 client. :return: An instance of this class. """ sqs_client = boto3.client('sqs') return cls(sqs_client) def set_queue_policy_for_topic(self, queue_arn: str, topic_arn: str, queue_url: str) -> bool: """ Set the queue policy to allow SNS to send messages to the queue. :param queue_arn: The ARN of the SQS queue. :param topic_arn: The ARN of the SNS topic. :param queue_url: The URL of the SQS queue. :return: True if successful. :raises ClientError: If setting the queue policy fails. """ try: # Create policy that allows SNS to send messages to the queue policy = { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "sns.amazonaws.com" }, "Action": "sqs:SendMessage", "Resource": queue_arn, "Condition": { "ArnEquals": { "aws:SourceArn": topic_arn } } } ] } self.sqs_client.set_queue_attributes( QueueUrl=queue_url, Attributes={ 'Policy': json.dumps(policy) } ) logger.info(f"Set queue policy for {queue_url} to allow messages from {topic_arn}") return True except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error setting queue policy: {error_code} - {e}") raise-
Per i dettagli sull'API, consulta SetQueueAttributes AWSSDK for Python (Boto3) API Reference.
-
Scenari
Il seguente esempio di codice mostra come creare un'applicazione di AWS Step Functions messaggistica che recupera i record dei messaggi da una tabella di database.
- SDK per Python (Boto3)
-
Mostra come usare AWS SDK per Python (Boto3) with per creare un'applicazione di messaggistica che AWS Step Functions recupera i record dei messaggi da una tabella Amazon DynamoDB e li invia con Amazon Simple Queue Service (Amazon SQS). La macchina a stati si integra con una AWS Lambda funzione per scansionare il database alla ricerca di messaggi non inviati.
Crea una macchina a stati che recuperi e aggiorni i record di messaggi da una tabella Amazon DynamoDB.
Aggiorna la definizione della macchina a stati per inviare messaggi anche ad Amazon Simple Queue Service (Amazon SQS).
Avvia e arresta l’esecuzione della macchina a stati.
Connettiti a Lambda, DynamoDB e Amazon SQS da una macchina a stati utilizzando le integrazioni di servizi.
Per il codice sorgente completo e le istruzioni su come configurarlo ed eseguirlo, vedi l'esempio completo su. GitHub
Servizi utilizzati in questo esempio
DynamoDB
Lambda
Amazon SQS
Step Functions
L’esempio di codice seguente mostra come esplorare l’output di Amazon Textract tramite un’applicazione interattiva.
- SDK per Python (Boto3)
-
Mostra come utilizzarlo AWS SDK per Python (Boto3) con Amazon Textract per rilevare elementi di testo, moduli e tabelle nell'immagine di un documento. L’immagine di input e l’output di Amazon Textract sono mostrati in un’applicazione Tkinter che consente di esplorare gli elementi rilevati.
Invia un’immagine del documento ad Amazon Textract ed esplora l’output degli elementi rilevati.
Invia immagini direttamente ad Amazon Textract o tramite un bucket Amazon Simple Storage Service (Amazon S3).
Utilizza asynchronous APIs per avviare un processo che pubblica una notifica su un argomento di Amazon Simple Notification Service (Amazon SNS) al termine del processo.
Esegue il polling di una coda Amazon Simple Queue Service (Amazon SQS) per un messaggio di completamento del processo e visualizza i risultati.
Per il codice sorgente completo e le istruzioni su come configurarlo ed eseguirlo, consulta l'esempio completo su. GitHub
Servizi utilizzati in questo esempio
Amazon Cognito Identity
Simple Storage Service (Amazon S3)
Amazon SNS
Amazon SQS
Amazon Textract
Gli esempi di codice seguenti mostrano come creare e pubblicare su un argomento FIFO Amazon SNS.
- SDK per Python (Boto3)
-
Nota
C'è altro su GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. Crea un argomento FIFO, sottoscrivi una coda FIFO di Amazon SQS all’argomento e pubblica un messaggio su un argomento Amazon SNS.
def usage_demo(): """Shows how to subscribe queues to a FIFO topic.""" print("-" * 88) print("Welcome to the `Subscribe queues to a FIFO topic` demo!") print("-" * 88) sns = boto3.resource("sns") sqs = boto3.resource("sqs") fifo_topic_wrapper = FifoTopicWrapper(sns) sns_wrapper = SnsWrapper(sns) prefix = "sqs-subscribe-demo-" queues = set() subscriptions = set() wholesale_queue = sqs.create_queue( QueueName=prefix + "wholesale.fifo", Attributes={ "MaximumMessageSize": str(4096), "ReceiveMessageWaitTimeSeconds": str(10), "VisibilityTimeout": str(300), "FifoQueue": str(True), "ContentBasedDeduplication": str(True), }, ) queues.add(wholesale_queue) print(f"Created FIFO queue with URL: {wholesale_queue.url}.") retail_queue = sqs.create_queue( QueueName=prefix + "retail.fifo", Attributes={ "MaximumMessageSize": str(4096), "ReceiveMessageWaitTimeSeconds": str(10), "VisibilityTimeout": str(300), "FifoQueue": str(True), "ContentBasedDeduplication": str(True), }, ) queues.add(retail_queue) print(f"Created FIFO queue with URL: {retail_queue.url}.") analytics_queue = sqs.create_queue(QueueName=prefix + "analytics", Attributes={}) queues.add(analytics_queue) print(f"Created standard queue with URL: {analytics_queue.url}.") topic = fifo_topic_wrapper.create_fifo_topic("price-updates-topic.fifo") print(f"Created FIFO topic: {topic.attributes['TopicArn']}.") for q in queues: fifo_topic_wrapper.add_access_policy(q, topic.attributes["TopicArn"]) print(f"Added access policies for topic: {topic.attributes['TopicArn']}.") for q in queues: sub = fifo_topic_wrapper.subscribe_queue_to_topic( topic, q.attributes["QueueArn"] ) subscriptions.add(sub) print(f"Subscribed queues to topic: {topic.attributes['TopicArn']}.") input("Press Enter to publish a message to the topic.") message_id = fifo_topic_wrapper.publish_price_update( topic, '{"product": 214, "price": 79.99}', "Consumables" ) print(f"Published price update with message ID: {message_id}.") # Clean up the subscriptions, queues, and topic. input("Press Enter to clean up resources.") for s in subscriptions: sns_wrapper.delete_subscription(s) sns_wrapper.delete_topic(topic) for q in queues: fifo_topic_wrapper.delete_queue(q) print(f"Deleted subscriptions, queues, and topic.") print("Thanks for watching!") print("-" * 88) class FifoTopicWrapper: """Encapsulates Amazon SNS FIFO topic and subscription functions.""" def __init__(self, sns_resource): """ :param sns_resource: A Boto3 Amazon SNS resource. """ self.sns_resource = sns_resource def create_fifo_topic(self, topic_name): """ Create a FIFO topic. Topic names must be made up of only uppercase and lowercase ASCII letters, numbers, underscores, and hyphens, and must be between 1 and 256 characters long. For a FIFO topic, the name must end with the .fifo suffix. :param topic_name: The name for the topic. :return: The new topic. """ try: topic = self.sns_resource.create_topic( Name=topic_name, Attributes={ "FifoTopic": str(True), "ContentBasedDeduplication": str(False), "FifoThroughputScope": "MessageGroup", }, ) logger.info("Created FIFO topic with name=%s.", topic_name) return topic except ClientError as error: logger.exception("Couldn't create topic with name=%s!", topic_name) raise error @staticmethod def add_access_policy(queue, topic_arn): """ Add the necessary access policy to a queue, so it can receive messages from a topic. :param queue: The queue resource. :param topic_arn: The ARN of the topic. :return: None. """ try: queue.set_attributes( Attributes={ "Policy": json.dumps( { "Version":"2012-10-17", "Statement": [ { "Sid": "test-sid", "Effect": "Allow", "Principal": {"AWS": "*"}, "Action": "SQS:SendMessage", "Resource": queue.attributes["QueueArn"], "Condition": { "ArnLike": {"aws:SourceArn": topic_arn} }, } ], } ) } ) logger.info("Added trust policy to the queue.") except ClientError as error: logger.exception("Couldn't add trust policy to the queue!") raise error @staticmethod def subscribe_queue_to_topic(topic, queue_arn): """ Subscribe a queue to a topic. :param topic: The topic resource. :param queue_arn: The ARN of the queue. :return: The subscription resource. """ try: subscription = topic.subscribe( Protocol="sqs", Endpoint=queue_arn, ) logger.info("The queue is subscribed to the topic.") return subscription except ClientError as error: logger.exception("Couldn't subscribe queue to topic!") raise error @staticmethod def publish_price_update(topic, payload, group_id): """ Compose and publish a message that updates the wholesale price. :param topic: The topic to publish to. :param payload: The message to publish. :param group_id: The group ID for the message. :return: The ID of the message. """ try: att_dict = {"business": {"DataType": "String", "StringValue": "wholesale"}} dedup_id = uuid.uuid4() response = topic.publish( Subject="Price Update", Message=payload, MessageAttributes=att_dict, MessageGroupId=group_id, MessageDeduplicationId=str(dedup_id), ) message_id = response["MessageId"] logger.info("Published message to topic %s.", topic.arn) except ClientError as error: logger.exception("Couldn't publish message to topic %s.", topic.arn) raise error return message_id @staticmethod def delete_queue(queue): """ Removes an SQS queue. When run against an AWS account, it can take up to 60 seconds before the queue is actually deleted. :param queue: The queue to delete. :return: None """ try: queue.delete() logger.info("Deleted queue with URL=%s.", queue.url) except ClientError as error: logger.exception("Couldn't delete queue with URL=%s!", queue.url) raise error-
Per informazioni dettagliate sull’API, consulta i seguenti argomenti nella documentazione di riferimento dell’API AWS SDK per Python (Boto3).
-
L’esempio di codice seguente mostra come rilevare persone e oggetti in un video con Amazon Rekognition.
- SDK per Python (Boto3)
-
Usa Amazon Rekognition per rilevare volti, oggetti e persone nei video avviando processi di rilevamento asincrono. Questo esempio, inoltre, configura Amazon Rekognition per notificare un argomento Amazon Simple Notification Service (Amazon SNS) al completamento dei processi e sottoscrive una coda Amazon Simple Queue Service (Amazon SQS) all’argomento. Quando la coda riceve un messaggio su un processo, questo viene recuperato e vengono restituiti i risultati.
Questo esempio è visualizzato al meglio su GitHub. Per il codice sorgente completo e le istruzioni su come configurarlo ed eseguirlo, vedi l'esempio completo su GitHub
. Servizi utilizzati in questo esempio
Amazon Rekognition
Simple Storage Service (Amazon S3)
Amazon SES
Amazon SNS
Amazon SQS
L’esempio di codice seguente mostra come:
Creazione di un argomento (FIFO o non FIFO).
Sottoscrizione di diverse code all’argomento con la possibilità di applicare un filtro.
Pubblicazione di un messaggio nell’argomento.
Esame delle code per i messaggi ricevuti.
- SDK per Python (Boto3)
-
Nota
C'è altro su GitHub. Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. Esegui uno scenario interattivo al prompt dei comandi.
class TopicsAndQueuesScenario: """Manages the Topics and Queues feature scenario.""" DASHES = "-" * 80 def __init__(self, sns_wrapper: SnsWrapper, sqs_wrapper: SqsWrapper) -> None: """ Initialize the Topics and Queues scenario. :param sns_wrapper: SnsWrapper instance for SNS operations. :param sqs_wrapper: SqsWrapper instance for SQS operations. """ self.sns_wrapper = sns_wrapper self.sqs_wrapper = sqs_wrapper # Scenario state self.use_fifo_topic = False self.use_content_based_deduplication = False self.topic_name = None self.topic_arn = None self.queue_count = 2 self.queue_urls = [] self.subscription_arns = [] self.tones = ["cheerful", "funny", "serious", "sincere"] def run_scenario(self) -> None: """Run the Topics and Queues feature scenario.""" print(self.DASHES) print("Welcome to messaging with topics and queues.") print(self.DASHES) print(f""" In this scenario, you will create an SNS topic and subscribe {self.queue_count} SQS queues to the topic. You can select from several options for configuring the topic and the subscriptions for the queues. You can then post to the topic and see the results in the queues. """) try: # Setup Phase print(self.DASHES) self._setup_topic() print(self.DASHES) self._setup_queues() print(self.DASHES) # Demonstration Phase self._publish_messages() print(self.DASHES) # Examination Phase self._poll_queues_for_messages() print(self.DASHES) # Cleanup Phase self._cleanup_resources() print(self.DASHES) except Exception as e: logger.error(f"Scenario failed: {e}") print(f"There was a problem with the scenario: {e}") print("\nInitiating cleanup...") try: self._cleanup_resources() except Exception as cleanup_error: logger.error(f"Error during cleanup: {cleanup_error}") print("Messaging with topics and queues scenario is complete.") print(self.DASHES) def _setup_topic(self) -> None: """Set up the SNS topic to be used with the queues.""" print("SNS topics can be configured as FIFO (First-In-First-Out).") print("FIFO topics deliver messages in order and support deduplication and message filtering.") print() self.use_fifo_topic = q.ask("Would you like to work with FIFO topics? (y/n): ", q.is_yesno) if self.use_fifo_topic: print(self.DASHES) self.topic_name = q.ask("Enter a name for your SNS topic: ", q.non_empty) print("Because you have selected a FIFO topic, '.fifo' must be appended to the topic name.") print() print(self.DASHES) print(""" Because you have chosen a FIFO topic, deduplication is supported. Deduplication IDs are either set in the message or automatically generated from content using a hash function. If a message is successfully published to an SNS FIFO topic, any message published and determined to have the same deduplication ID, within the five-minute deduplication interval, is accepted but not delivered. For more information about deduplication, see https://docs.aws.amazon.com/sns/latest/dg/fifo-message-dedup.html. """) self.use_content_based_deduplication = q.ask( "Use content-based deduplication instead of entering a deduplication ID? (y/n): ", q.is_yesno ) else: self.topic_name = q.ask("Enter a name for your SNS topic: ", q.non_empty) print(self.DASHES) # Create the topic self.topic_arn = self.sns_wrapper.create_topic( self.topic_name, self.use_fifo_topic, self.use_content_based_deduplication ) print(f"Your new topic with the name {self.topic_name}") print(f" and Amazon Resource Name (ARN) {self.topic_arn}") print(f" has been created.") print() def _setup_queues(self) -> None: """Set up the SQS queues and subscribe them to the topic.""" print(f"Now you will create {self.queue_count} Amazon Simple Queue Service (Amazon SQS) queues to subscribe to the topic.") for i in range(self.queue_count): queue_name = q.ask(f"Enter a name for SQS queue #{i+1}: ", q.non_empty) if self.use_fifo_topic and i == 0: print("Because you have selected a FIFO topic, '.fifo' must be appended to the queue name.") # Create the queue queue_url = self.sqs_wrapper.create_queue(queue_name, self.use_fifo_topic) self.queue_urls.append(queue_url) print(f"Your new queue with the name {queue_name}") print(f" and queue URL {queue_url}") print(f" has been created.") print() if i == 0: print("The queue URL is used to retrieve the queue ARN,") print("which is used to create a subscription.") print(self.DASHES) # Get queue ARN queue_arn = self.sqs_wrapper.get_queue_arn(queue_url) if i == 0: print("An AWS Identity and Access Management (IAM) policy must be attached to an SQS queue,") print("enabling it to receive messages from an SNS topic.") # Set queue policy to allow SNS to send messages self.sqs_wrapper.set_queue_policy_for_topic(queue_arn, self.topic_arn, queue_url) # Set up message filtering if using FIFO subscription_arn = self._setup_subscription_with_filter(i, queue_arn, queue_name) self.subscription_arns.append(subscription_arn) def _setup_subscription_with_filter(self, queue_index: int, queue_arn: str, queue_name: str) -> str: """Set up subscription with optional message filtering.""" filter_policy = None if self.use_fifo_topic: print(self.DASHES) if queue_index == 0: print("Subscriptions to a FIFO topic can have filters.") print("If you add a filter to this subscription, then only the filtered messages") print("will be received in the queue.") print() print("For information about message filtering,") print("see https://docs.aws.amazon.com/sns/latest/dg/sns-message-filtering.html") print() print("For this example, you can filter messages by a TONE attribute.") use_filter = q.ask(f"Filter messages for {queue_name}'s subscription to the topic? (y/n): ", q.is_yesno) if use_filter: filter_policy = self._create_filter_policy() subscription_arn = self.sns_wrapper.subscribe_queue_to_topic( self.topic_arn, queue_arn, filter_policy ) print(f"The queue {queue_name} has been subscribed to the topic {self.topic_name}") print(f" with the subscription ARN {subscription_arn}") return subscription_arn def _create_filter_policy(self) -> str: """Create a message filter policy based on user selections.""" print(self.DASHES) print("You can filter messages by one or more of the following TONE attributes.") filter_selections = [] selection_number = 0 while True: print("Enter a number to add a TONE filter, or enter 0 to stop adding filters.") for i, tone in enumerate(self.tones, 1): print(f" {i}. {tone}") selection = q.ask("Your choice: ", q.is_int, q.in_range(0, len(self.tones))) if selection == 0: break elif selection > 0 and self.tones[selection - 1] not in filter_selections: filter_selections.append(self.tones[selection - 1]) print(f"Added '{self.tones[selection - 1]}' to filter list.") if filter_selections: filters = {"tone": filter_selections} return json.dumps(filters) return None def _publish_messages(self) -> None: """Publish messages to the topic with various options.""" print("Now we can publish messages.") keep_sending = True while keep_sending: print() message = q.ask("Enter a message to publish: ", q.non_empty) message_group_id = None deduplication_id = None tone_attribute = None if self.use_fifo_topic: print("Because you are using a FIFO topic, you must set a message group ID.") print("All messages within the same group will be received in the order they were published.") print() message_group_id = q.ask("Enter a message group ID for this message: ", q.non_empty) if not self.use_content_based_deduplication: print("Because you are not using content-based deduplication,") print("you must enter a deduplication ID.") deduplication_id = q.ask("Enter a deduplication ID for this message: ", q.non_empty) # Ask about tone attribute add_attribute = q.ask("Add an attribute to this message? (y/n): ", q.is_yesno) if add_attribute: print("Enter a number for an attribute:") for i, tone in enumerate(self.tones, 1): print(f" {i}. {tone}") selection = q.ask("Your choice: ", q.is_int, q.in_range(1, len(self.tones))) if 1 <= selection <= len(self.tones): tone_attribute = self.tones[selection - 1] # Publish the message message_id = self.sns_wrapper.publish_message( self.topic_arn, message, tone_attribute, deduplication_id, message_group_id ) print(f"Message published with ID: {message_id}") keep_sending = q.ask("Send another message? (y/n): ", q.is_yesno) def _poll_queues_for_messages(self) -> None: """Poll all queues for messages and display results.""" for i, queue_url in enumerate(self.queue_urls): print(f"Polling queue #{i+1} at {queue_url} for messages...") q.ask("Press Enter to continue...") messages = self._poll_queue_for_messages(queue_url) if messages: print(f"{len(messages)} message(s) were received by queue #{i+1}") for j, message in enumerate(messages, 1): print(f" Message {j}:") # Parse the SNS message body to get the actual message try: sns_message = json.loads(message['Body']) actual_message = sns_message.get('Message', message['Body']) print(f" {actual_message}") except (json.JSONDecodeError, KeyError): print(f" {message['Body']}") # Delete the messages self.sqs_wrapper.delete_messages(queue_url, messages) print(f"Messages deleted from queue #{i+1}") else: print(f"No messages received by queue #{i+1}") print(self.DASHES) def _poll_queue_for_messages(self, queue_url: str) -> List[Dict[str, Any]]: """Poll a single queue for messages.""" all_messages = [] max_polls = 3 # Limit polling to avoid infinite loops for poll_count in range(max_polls): messages = self.sqs_wrapper.receive_messages(queue_url, 10) if messages: all_messages.extend(messages) print(f" Received {len(messages)} messages in poll {poll_count + 1}") # Small delay between polls time.sleep(1) else: print(f" No messages in poll {poll_count + 1}") break return all_messages def _cleanup_resources(self) -> None: """Clean up all resources created during the scenario.""" print("Cleaning up resources...") # Delete queues for i, queue_url in enumerate(self.queue_urls): if queue_url: delete_queue = q.ask(f"Delete queue #{i+1} with URL {queue_url}? (y/n): ", q.is_yesno) if delete_queue: try: self.sqs_wrapper.delete_queue(queue_url) print(f"Deleted queue #{i+1}") except Exception as e: print(f"Error deleting queue #{i+1}: {e}") # Unsubscribe from topic for i, subscription_arn in enumerate(self.subscription_arns): if subscription_arn: try: self.sns_wrapper.unsubscribe(subscription_arn) print(f"Unsubscribed subscription #{i+1}") except Exception as e: print(f"Error unsubscribing #{i+1}: {e}") # Delete topic if self.topic_arn: delete_topic = q.ask(f"Delete topic {self.topic_name}? (y/n): ", q.is_yesno) if delete_topic: try: self.sns_wrapper.delete_topic(self.topic_arn) print(f"Deleted topic {self.topic_name}") except Exception as e: print(f"Error deleting topic: {e}") print("Resource cleanup complete.")Crea classi che racchiudono le operazioni di Amazon SNS e Amazon SQS da utilizzare nello scenario.
class SnsWrapper: """Wrapper class for managing Amazon SNS operations.""" def __init__(self, sns_client: Any) -> None: """ Initialize the SnsWrapper. :param sns_client: A Boto3 Amazon SNS client. """ self.sns_client = sns_client @classmethod def from_client(cls) -> 'SnsWrapper': """ Create an SnsWrapper instance using a default boto3 client. :return: An instance of this class. """ sns_client = boto3.client('sns') return cls(sns_client) def create_topic( self, topic_name: str, is_fifo: bool = False, content_based_deduplication: bool = False ) -> str: """ Create an SNS topic. :param topic_name: The name of the topic to create. :param is_fifo: Whether to create a FIFO topic. :param content_based_deduplication: Whether to use content-based deduplication for FIFO topics. :return: The ARN of the created topic. :raises ClientError: If the topic creation fails. """ try: # Add .fifo suffix for FIFO topics if is_fifo and not topic_name.endswith('.fifo'): topic_name += '.fifo' attributes = {} if is_fifo: attributes['FifoTopic'] = 'true' if content_based_deduplication: attributes['ContentBasedDeduplication'] = 'true' response = self.sns_client.create_topic( Name=topic_name, Attributes=attributes ) topic_arn = response['TopicArn'] logger.info(f"Created topic: {topic_name} with ARN: {topic_arn}") return topic_arn except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error creating topic {topic_name}: {error_code} - {e}") raise def subscribe_queue_to_topic( self, topic_arn: str, queue_arn: str, filter_policy: Optional[str] = None ) -> str: """ Subscribe an SQS queue to an SNS topic. :param topic_arn: The ARN of the SNS topic. :param queue_arn: The ARN of the SQS queue. :param filter_policy: Optional JSON filter policy for message filtering. :return: The ARN of the subscription. :raises ClientError: If the subscription fails. """ try: attributes = {} if filter_policy: attributes['FilterPolicy'] = filter_policy response = self.sns_client.subscribe( TopicArn=topic_arn, Protocol='sqs', Endpoint=queue_arn, Attributes=attributes ) subscription_arn = response['SubscriptionArn'] logger.info(f"Subscribed queue {queue_arn} to topic {topic_arn}") return subscription_arn except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error subscribing queue to topic: {error_code} - {e}") raise def publish_message( self, topic_arn: str, message: str, tone_attribute: Optional[str] = None, deduplication_id: Optional[str] = None, message_group_id: Optional[str] = None ) -> str: """ Publish a message to an SNS topic. :param topic_arn: The ARN of the SNS topic. :param message: The message content to publish. :param tone_attribute: Optional tone attribute for message filtering. :param deduplication_id: Optional deduplication ID for FIFO topics. :param message_group_id: Optional message group ID for FIFO topics. :return: The message ID of the published message. :raises ClientError: If the message publication fails. """ try: publish_args = { 'TopicArn': topic_arn, 'Message': message } # Add message attributes if tone is specified if tone_attribute: publish_args['MessageAttributes'] = { 'tone': { 'DataType': 'String', 'StringValue': tone_attribute } } # Add FIFO-specific parameters if message_group_id: publish_args['MessageGroupId'] = message_group_id if deduplication_id: publish_args['MessageDeduplicationId'] = deduplication_id response = self.sns_client.publish(**publish_args) message_id = response['MessageId'] logger.info(f"Published message to topic {topic_arn} with ID: {message_id}") return message_id except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error publishing message to topic: {error_code} - {e}") raise def unsubscribe(self, subscription_arn: str) -> bool: """ Unsubscribe from an SNS topic. :param subscription_arn: The ARN of the subscription to remove. :return: True if successful. :raises ClientError: If the unsubscribe operation fails. """ try: self.sns_client.unsubscribe(SubscriptionArn=subscription_arn) logger.info(f"Unsubscribed: {subscription_arn}") return True except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') if error_code == 'NotFound': logger.warning(f"Subscription not found: {subscription_arn}") return True # Already unsubscribed else: logger.error(f"Error unsubscribing: {error_code} - {e}") raise def delete_topic(self, topic_arn: str) -> bool: """ Delete an SNS topic. :param topic_arn: The ARN of the topic to delete. :return: True if successful. :raises ClientError: If the topic deletion fails. """ try: self.sns_client.delete_topic(TopicArn=topic_arn) logger.info(f"Deleted topic: {topic_arn}") return True except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') if error_code == 'NotFound': logger.warning(f"Topic not found: {topic_arn}") return True # Already deleted else: logger.error(f"Error deleting topic: {error_code} - {e}") raise def list_topics(self) -> list: """ List all SNS topics in the account using pagination. :return: List of topic ARNs. :raises ClientError: If listing topics fails. """ try: topics = [] paginator = self.sns_client.get_paginator('list_topics') for page in paginator.paginate(): topics.extend([topic['TopicArn'] for topic in page.get('Topics', [])]) logger.info(f"Found {len(topics)} topics") return topics except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') if error_code == 'AuthorizationError': logger.error("Authorization error listing topics - check IAM permissions") else: logger.error(f"Error listing topics: {error_code} - {e}") raise class SqsWrapper: """Wrapper class for managing Amazon SQS operations.""" def __init__(self, sqs_client: Any) -> None: """ Initialize the SqsWrapper. :param sqs_client: A Boto3 Amazon SQS client. """ self.sqs_client = sqs_client @classmethod def from_client(cls) -> 'SqsWrapper': """ Create an SqsWrapper instance using a default boto3 client. :return: An instance of this class. """ sqs_client = boto3.client('sqs') return cls(sqs_client) def create_queue(self, queue_name: str, is_fifo: bool = False) -> str: """ Create an SQS queue. :param queue_name: The name of the queue to create. :param is_fifo: Whether to create a FIFO queue. :return: The URL of the created queue. :raises ClientError: If the queue creation fails. """ try: # Add .fifo suffix for FIFO queues if is_fifo and not queue_name.endswith('.fifo'): queue_name += '.fifo' attributes = {} if is_fifo: attributes['FifoQueue'] = 'true' response = self.sqs_client.create_queue( QueueName=queue_name, Attributes=attributes ) queue_url = response['QueueUrl'] logger.info(f"Created queue: {queue_name} with URL: {queue_url}") return queue_url except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error creating queue {queue_name}: {error_code} - {e}") raise def get_queue_arn(self, queue_url: str) -> str: """ Get the ARN of an SQS queue. :param queue_url: The URL of the queue. :return: The ARN of the queue. :raises ClientError: If getting queue attributes fails. """ try: response = self.sqs_client.get_queue_attributes( QueueUrl=queue_url, AttributeNames=['QueueArn'] ) queue_arn = response['Attributes']['QueueArn'] logger.info(f"Queue ARN for {queue_url}: {queue_arn}") return queue_arn except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error getting queue ARN: {error_code} - {e}") raise def set_queue_policy_for_topic(self, queue_arn: str, topic_arn: str, queue_url: str) -> bool: """ Set the queue policy to allow SNS to send messages to the queue. :param queue_arn: The ARN of the SQS queue. :param topic_arn: The ARN of the SNS topic. :param queue_url: The URL of the SQS queue. :return: True if successful. :raises ClientError: If setting the queue policy fails. """ try: # Create policy that allows SNS to send messages to the queue policy = { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "sns.amazonaws.com" }, "Action": "sqs:SendMessage", "Resource": queue_arn, "Condition": { "ArnEquals": { "aws:SourceArn": topic_arn } } } ] } self.sqs_client.set_queue_attributes( QueueUrl=queue_url, Attributes={ 'Policy': json.dumps(policy) } ) logger.info(f"Set queue policy for {queue_url} to allow messages from {topic_arn}") return True except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error setting queue policy: {error_code} - {e}") raise def receive_messages(self, queue_url: str, max_messages: int = 10) -> List[Dict[str, Any]]: """ Receive messages from an SQS queue. :param queue_url: The URL of the queue to receive messages from. :param max_messages: Maximum number of messages to receive (1-10). :return: List of received messages. :raises ClientError: If receiving messages fails. """ try: # Ensure max_messages is within valid range max_messages = max(1, min(10, max_messages)) response = self.sqs_client.receive_message( QueueUrl=queue_url, MaxNumberOfMessages=max_messages, WaitTimeSeconds=2, # Short polling MessageAttributeNames=['All'] ) messages = response.get('Messages', []) logger.info(f"Received {len(messages)} messages from {queue_url}") return messages except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error receiving messages: {error_code} - {e}") raise def delete_messages(self, queue_url: str, messages: List[Dict[str, Any]]) -> bool: """ Delete messages from an SQS queue in batches. :param queue_url: The URL of the queue. :param messages: List of messages to delete. :return: True if successful. :raises ClientError: If deleting messages fails. """ try: if not messages: return True # Build delete entries for batch delete delete_entries = [] for i, message in enumerate(messages): delete_entries.append({ 'Id': str(i), 'ReceiptHandle': message['ReceiptHandle'] }) # Delete messages in batches of 10 (SQS limit) batch_size = 10 for i in range(0, len(delete_entries), batch_size): batch = delete_entries[i:i + batch_size] response = self.sqs_client.delete_message_batch( QueueUrl=queue_url, Entries=batch ) # Check for failures if 'Failed' in response and response['Failed']: for failed in response['Failed']: logger.warning(f"Failed to delete message: {failed}") logger.info(f"Deleted {len(messages)} messages from {queue_url}") return True except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error deleting messages: {error_code} - {e}") raise def delete_queue(self, queue_url: str) -> bool: """ Delete an SQS queue. :param queue_url: The URL of the queue to delete. :return: True if successful. :raises ClientError: If the queue deletion fails. """ try: self.sqs_client.delete_queue(QueueUrl=queue_url) logger.info(f"Deleted queue: {queue_url}") return True except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') if error_code == 'AWS.SimpleQueueService.NonExistentQueue': logger.warning(f"Queue not found: {queue_url}") return True # Already deleted else: logger.error(f"Error deleting queue: {error_code} - {e}") raise def list_queues(self, queue_name_prefix: Optional[str] = None) -> List[str]: """ List all SQS queues in the account using pagination. :param queue_name_prefix: Optional prefix to filter queue names. :return: List of queue URLs. :raises ClientError: If listing queues fails. """ try: queue_urls = [] paginator = self.sqs_client.get_paginator('list_queues') page_params = {} if queue_name_prefix: page_params['QueueNamePrefix'] = queue_name_prefix for page in paginator.paginate(**page_params): queue_urls.extend(page.get('QueueUrls', [])) logger.info(f"Found {len(queue_urls)} queues") return queue_urls except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') if error_code == 'AccessDenied': logger.error("Access denied listing queues - check IAM permissions") else: logger.error(f"Error listing queues: {error_code} - {e}") raise def send_message(self, queue_url: str, message_body: str, **kwargs) -> str: """ Send a message to an SQS queue. :param queue_url: The URL of the queue. :param message_body: The message content. :param kwargs: Additional message parameters (DelaySeconds, MessageAttributes, etc.). :return: The message ID. :raises ClientError: If sending the message fails. """ try: send_params = { 'QueueUrl': queue_url, 'MessageBody': message_body, **kwargs } response = self.sqs_client.send_message(**send_params) message_id = response['MessageId'] logger.info(f"Sent message to {queue_url} with ID: {message_id}") return message_id except ClientError as e: error_code = e.response.get('Error', {}).get('Code', 'Unknown') logger.error(f"Error sending message: {error_code} - {e}") raise-
Per informazioni dettagliate sull’API, consulta i seguenti argomenti nella documentazione di riferimento dell’API AWS SDK per Python (Boto3).
-
L’esempio di codice seguente mostra come:
Creare una coda Amazon SQS.
Inviare batch di messaggi alla coda.
Ricevere batch di messaggi dalla coda.
Eliminare batch di messaggi dalla coda.
- SDK per Python (Boto3)
-
Nota
C'è altro da fare. GitHub Trova l'esempio completo e scopri di più sulla configurazione e l'esecuzione nel Repository di esempi di codice AWS
. Crea funzioni per il wrapping delle funzioni dei messaggi Amazon SQS.
import logging import sys import boto3 from botocore.exceptions import ClientError import queue_wrapper logger = logging.getLogger(__name__) sqs = boto3.resource("sqs") def send_messages(queue, messages): """ Send a batch of messages in a single request to an SQS queue. This request may return overall success even when some messages were not sent. The caller must inspect the Successful and Failed lists in the response and resend any failed messages. :param queue: The queue to receive the messages. :param messages: The messages to send to the queue. These are simplified to contain only the message body and attributes. :return: The response from SQS that contains the list of successful and failed messages. """ try: entries = [ { "Id": str(ind), "MessageBody": msg["body"], "MessageAttributes": msg["attributes"], } for ind, msg in enumerate(messages) ] response = queue.send_messages(Entries=entries) if "Successful" in response: for msg_meta in response["Successful"]: logger.info( "Message sent: %s: %s", msg_meta["MessageId"], messages[int(msg_meta["Id"])]["body"], ) if "Failed" in response: for msg_meta in response["Failed"]: logger.warning( "Failed to send: %s: %s", msg_meta["MessageId"], messages[int(msg_meta["Id"])]["body"], ) except ClientError as error: logger.exception("Send messages failed to queue: %s", queue) raise error else: return response def receive_messages(queue, max_number, wait_time): """ Receive a batch of messages in a single request from an SQS queue. :param queue: The queue from which to receive messages. :param max_number: The maximum number of messages to receive. The actual number of messages received might be less. :param wait_time: The maximum time to wait (in seconds) before returning. When this number is greater than zero, long polling is used. This can result in reduced costs and fewer false empty responses. :return: The list of Message objects received. These each contain the body of the message and metadata and custom attributes. """ try: messages = queue.receive_messages( MessageAttributeNames=["All"], MaxNumberOfMessages=max_number, WaitTimeSeconds=wait_time, ) for msg in messages: logger.info("Received message: %s: %s", msg.message_id, msg.body) except ClientError as error: logger.exception("Couldn't receive messages from queue: %s", queue) raise error else: return messages def delete_messages(queue, messages): """ Delete a batch of messages from a queue in a single request. :param queue: The queue from which to delete the messages. :param messages: The list of messages to delete. :return: The response from SQS that contains the list of successful and failed message deletions. """ try: entries = [ {"Id": str(ind), "ReceiptHandle": msg.receipt_handle} for ind, msg in enumerate(messages) ] response = queue.delete_messages(Entries=entries) if "Successful" in response: for msg_meta in response["Successful"]: logger.info("Deleted %s", messages[int(msg_meta["Id"])].receipt_handle) if "Failed" in response: for msg_meta in response["Failed"]: logger.warning( "Could not delete %s", messages[int(msg_meta["Id"])].receipt_handle ) except ClientError: logger.exception("Couldn't delete messages from queue %s", queue) else: return responseUtilizza le funzioni wrapper per inviare e ricevere messaggi in batch.
def usage_demo(): """ Shows how to: * Read the lines from this Python file and send the lines in batches of 10 as messages to a queue. * Receive the messages in batches until the queue is empty. * Reassemble the lines of the file and verify they match the original file. """ def pack_message(msg_path, msg_body, msg_line): return { "body": msg_body, "attributes": { "path": {"StringValue": msg_path, "DataType": "String"}, "line": {"StringValue": str(msg_line), "DataType": "String"}, }, } def unpack_message(msg): return ( msg.message_attributes["path"]["StringValue"], msg.body, int(msg.message_attributes["line"]["StringValue"]), ) print("-" * 88) print("Welcome to the Amazon Simple Queue Service (Amazon SQS) demo!") print("-" * 88) queue = queue_wrapper.create_queue("sqs-usage-demo-message-wrapper") with open(__file__) as file: lines = file.readlines() line = 0 batch_size = 10 received_lines = [None] * len(lines) print(f"Sending file lines in batches of {batch_size} as messages.") while line < len(lines): messages = [ pack_message(__file__, lines[index], index) for index in range(line, min(line + batch_size, len(lines))) ] line = line + batch_size send_messages(queue, messages) print(".", end="") sys.stdout.flush() print(f"Done. Sent {len(lines) - 1} messages.") print(f"Receiving, handling, and deleting messages in batches of {batch_size}.") more_messages = True while more_messages: received_messages = receive_messages(queue, batch_size, 2) print(".", end="") sys.stdout.flush() for message in received_messages: path, body, line = unpack_message(message) received_lines[line] = body if received_messages: delete_messages(queue, received_messages) else: more_messages = False print("Done.") if all([lines[index] == received_lines[index] for index in range(len(lines))]): print(f"Successfully reassembled all file lines!") else: print(f"Uh oh, some lines were missed!") queue.delete() print("Thanks for watching!") print("-" * 88)-
Per informazioni dettagliate sull’API, consulta i seguenti argomenti nella documentazione di riferimento dell’API AWS SDK per Python (Boto3).
-
Esempi serverless
L’esempio di codice seguente mostra come implementare una funzione Lambda che riceve un evento attivato dalla ricezione di messaggi da una coda SQS. La funzione recupera i messaggi dal parametro dell’evento e registra il contenuto di ogni messaggio.
- SDK per Python (Boto3)
-
Nota
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di Esempi serverless
. Utilizzo di un evento SQS con Lambda tramite Python.
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 def lambda_handler(event, context): for message in event['Records']: process_message(message) print("done") def process_message(message): try: print(f"Processed message {message['body']}") # TODO: Do interesting work based on the new message except Exception as err: print("An error occurred") raise err
L’esempio di codice seguente mostra come implementare una risposta batch parziale per funzioni Lambda che ricevono eventi da una coda SQS. La funzione riporta gli errori degli elementi batch nella risposta, segnalando a Lambda di riprovare tali messaggi in un secondo momento.
- SDK per Python (Boto3)
-
Nota
C'è dell'altro GitHub. Trova l’esempio completo e scopri come eseguire la configurazione e l’esecuzione nel repository di Esempi serverless
. Segnalazione di errori di elementi batch di SQS con Lambda tramite Python.
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 def lambda_handler(event, context): if event: batch_item_failures = [] sqs_batch_response = {} for record in event["Records"]: try: print(f"Processed message: {record['body']}") except Exception as e: batch_item_failures.append({"itemIdentifier": record['messageId']}) sqs_batch_response["batchItemFailures"] = batch_item_failures return sqs_batch_response
