Weitere AWS-SDK-Beispiele sind im GitHub-Repository Beispiele für AWS Doc SDKs
Beispiele für Amazon SQS unter Verwendung von SDK für Python (Boto3)
Die folgenden Codebeispiele zeigen, wie Sie Aktionen durchführen und gängige Szenarien implementieren, indem Sie AWS SDK für Python (Boto3) mit Amazon SQS nutzen.
Aktionen sind Codeauszüge aus größeren Programmen und müssen im Kontext ausgeführt werden. Während Aktionen Ihnen zeigen, wie Sie einzelne Servicefunktionen aufrufen, können Sie Aktionen im Kontext der zugehörigen Szenarien anzeigen.
Szenarien sind Codebeispiele, die Ihnen zeigen, wie Sie bestimmte Aufgaben ausführen, indem Sie mehrere Funktionen innerhalb eines Services aufrufen oder mit anderen AWS-Services kombinieren.
Jedes Beispiel enthält einen Link zum vollständigen Quellcode, wo Sie Anweisungen zum Einrichten und Ausführen des Codes im Kodex finden.
Aktionen
Die folgenden Codebeispiele zeigen, wie CreateQueue verwendet wird.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. 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 queue-
Weitere API-Informationen finden Sie unter CreateQueue in der API-Referenz zum AWS-SDK für Python (Boto3).
-
Die folgenden Codebeispiele zeigen, wie DeleteMessage verwendet wird.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. 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-
Weitere API-Informationen finden Sie unter DeleteMessage in der API-Referenz zum AWS-SDK für Python (Boto3).
-
Die folgenden Codebeispiele zeigen, wie DeleteMessageBatch verwendet wird.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. 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 response-
Weitere API-Informationen finden Sie unter DeleteMessageBatch in der API-Referenz zu AWS-SDK für Python (Boto3).
-
Die folgenden Codebeispiele zeigen, wie DeleteQueue verwendet wird.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. 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 error-
Weitere API-Informationen finden Sie unter DeleteQueue in der API-Referenz zum AWS-SDK für Python (Boto3).
-
Die folgenden Codebeispiele zeigen, wie GetQueueUrl verwendet wird.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. 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-
Weitere API-Informationen finden Sie unter GetQueueUrl in der API-Referenz zum AWS-SDK für Python (Boto3).
-
Die folgenden Codebeispiele zeigen, wie ListQueues verwendet wird.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. 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-
Weitere API-Informationen finden Sie unter ListQueues in der API-Referenz zum AWS-SDK für Python (Boto3).
-
Die folgenden Codebeispiele zeigen, wie ReceiveMessage verwendet wird.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. 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-
Weitere API-Informationen finden Sie unter ReceiveMessage in der API-Referenz zum AWS-SDK für Python (Boto3).
-
Die folgenden Codebeispiele zeigen, wie SendMessage verwendet wird.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. 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-
Weitere API-Informationen finden Sie unter SendMessage in der API-Referenz zum AWS-SDK für Python (Boto3).
-
Die folgenden Codebeispiele zeigen, wie SendMessageBatch verwendet wird.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. 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-
Weitere API-Informationen finden Sie unter SendMessageBatch in der API-Referenz zum AWS-SDK für Python (Boto3).
-
Szenarien
Das folgende Codebeispiel veranschaulicht, wie eine AWS Step Functions-Messenger-Anwendung erstellt wird, die Nachrichtendatensätze aus einer Datenbanktabelle abruft.
- SDK für Python (Boto3)
-
Veranschaulicht die Verwendung von AWS SDK für Python (Boto3) mit AWS Step Functions, um eine Messenger-Anwendung zu erstellen, die Nachrichtendatensätze aus einer Amazon-DynamoDB-Tabelle abruft und mit Amazon Simple Queue Service (Amazon SQS) sendet. Der Zustandsautomat kann in Verbindung mit einer AWS Lambda-Funktion verwendet werden, um die Datenbank auf nicht gesendete Nachrichten zu scannen.
Erstellen Sie einen Zustandsautomaten, der Nachrichtendatensätze aus einer Amazon-DynamoDB-Tabelle abruft und aktualisiert.
Aktualisieren Sie die Definition des Zustandsautomaten, um auch Nachrichten an Amazon Simple Queue Service (Amazon SQS) zu senden.
Starten und stoppen Sie Ausführungen des Zustandsautomaten.
Stellen Sie vom Zustandsautomaten aus über Serviceintegrationen eine Verbindung zu Lambda, DynamoDB und Amazon SQS her.
Den kompletten Quellcode und Anweisungen zum Einrichten und Ausführen finden Sie im vollständigen Beispiel unter GitHub
. In diesem Beispiel verwendete Services
DynamoDB
Lambda
Amazon SQS
Step Functions
Die folgenden Codebeispiele zeigen, wie die Amazon-Textract-Ausgabe in einer interaktiven Anwendung untersucht werden kann.
- SDK für Python (Boto3)
-
Zeigt, wie man AWS SDK für Python (Boto3) mit Amazon Textract verwendet, um Text-, Formular- und Tabellenelemente in einem Dokument-Image zu erkennen. Das Eingabe-Image und die Amazon-Textract-Ausgabe werden in einer Tkinter-Anwendung angezeigt, mit der Sie die erkannten Elemente untersuchen können.
Senden Sie ein Dokument-Image an Amazon Textract und untersuchen Sie die Ausgabe erkannter Elemente.
Senden Sie Images direkt an Amazon Textract oder über einen Amazon Simple Storage Service (Amazon S3)-Bucket.
Verwenden Sie asynchrone APIs, um einen Auftrag zu starten, der eine Benachrichtigung an ein Amazon Simple Notification Service (Amazon SNS)-Thema veröffentlicht.
Stellen Sie eine Amazon Simple Queue Service (Amazon SQS)-Warteschlange ab, um eine Meldung zum Abschluss des Auftrags zu erhalten.
Den kompletten Quellcode und Anweisungen zum Einrichten und Ausführen finden Sie im vollständigen Beispiel unter GitHub
. In diesem Beispiel verwendete Services
Amazon Cognito Identity
Amazon S3
Amazon SNS
Amazon SQS
Amazon Textract
Die folgenden Code-Beispiele zeigen, wie man ein Amazon-SNS-Thema erstellt.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. Erstellen Sie ein Amazon-SNS-FIFO-Thema, abonnieren Sie eine Amazon-SQS-FIFO- und eine Standard-Warteschlange für das Thema und veröffentlichen Sie eine Nachricht zu dem Thema.
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-
Weitere API-Informationen finden Sie in den folgenden Themen der API-Referenz zum AWS-SDK für Python (Boto3).
-
Das folgende Codebeispiel zeigt, wie Sie Personen und Objekte in einem Video mit Amazon Rekognition erkennen.
- SDK für Python (Boto3)
-
Verwenden Sie Amazon Rekognition, um Gesichter, Objekte und Personen in Videos zu erkennen, indem Sie asynchrone Erkennungsaufträge starten. In diesem Beispiel wird Amazon Rekognition auch so konfiguriert, dass es ein Amazon Simple Notification Service (Amazon SNS)-Thema benachrichtigt, wenn Aufträge abgeschlossen sind, und eine Amazon Simple Queue Service (Amazon SQS)-Warteschlange bei dem Thema abonniert. Wenn die Warteschlange eine Meldung über einen Job erhält, wird der Job abgerufen und die Ergebnisse werden ausgegeben.
Dieses Beispiel wird am besten auf GitHub angesehen. Den kompletten Quellcode und Anweisungen zum Einrichten und Ausführen finden Sie im vollständigen Beispiel unter GitHub
. In diesem Beispiel verwendete Services
Amazon Rekognition
Amazon S3
Amazon SES
Amazon SNS
Amazon SQS
Wie das aussehen kann, sehen Sie am nachfolgenden Beispielcode:
Erstellen einer Amazon SQS-Warteschlange
Senden Sie Stapel von Nachrichten an die Warteschlange.
Empfangen Sie Stapel von Nachrichten aus der Warteschlange.
Löschen Sie Stapel von Nachrichten aus der Warteschlange.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das AWS-Codebeispiel-Repository
einrichten und ausführen. Erstellen Sie Funktionen, um Amazon-SQS-Nachrichtenfunktionen zu umschließen.
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 responseVerwenden Sie die Wrapper-Funktionen, um Nachrichten stapelweise zu senden und zu empfangen.
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)-
Weitere API-Informationen finden Sie in den folgenden Themen der API-Referenz zum AWS-SDK für Python (Boto3).
-
Serverless-Beispiele
Das folgende Codebeispiel zeigt, wie eine Lambda-Funktion implementiert wird, die ein Ereignis empfängt, das durch den Empfang von Nachrichten aus einer SQS-Warteschlange ausgelöst wird. Die Funktion ruft die Nachrichten aus dem Ereignisparameter ab und protokolliert den Inhalt jeder Nachricht.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Das vollständige Beispiel sowie eine Anleitung zum Einrichten und Ausführen finden Sie im Repository mit Serverless-Beispielen
. Nutzen eines SQS-Ereignisses mit Lambda unter Verwendung von 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
Das folgende Codebeispiel zeigt, wie eine teilweise Batch-Antwort für Lambda-Funktionen implementiert wird, die Ereignisse aus einer SQS-Warteschlange empfangen. Die Funktion meldet die Batch-Elementfehler in der Antwort und signalisiert Lambda, diese Nachrichten später erneut zu versuchen.
- SDK für Python (Boto3)
-
Anmerkung
Auf GitHub finden Sie noch mehr. Das vollständige Beispiel sowie eine Anleitung zum Einrichten und Ausführen finden Sie im Repository mit Serverless-Beispielen
. Melden von Fehlern bei SQS-Batchelementen mit Lambda unter Verwendung von 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: # process message except Exception as e: batch_item_failures.append({"itemIdentifier": record['messageId']}) sqs_batch_response["batchItemFailures"] = batch_item_failures return sqs_batch_response
