Doc AWS SDK 예제 GitHub 리포지토리에서 더 많은 SDK 예제를 사용할 수 있습니다. [AWS](https://github.com/awsdocs/aws-doc-sdk-examples)
기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.
# SDK for Python (Boto3)을 사용한 Amazon SQS 예제
다음 코드 예제에서는 AWS SDK for Python (Boto3) Amazon SQS를 사용하여 작업을 수행하고 일반적인 시나리오를 구현하는 방법을 보여줍니다.
*작업*은 대규모 프로그램에서 발췌한 코드이며 컨텍스트에 맞춰 실행해야 합니다. 작업은 개별 서비스 함수를 직접 호출하는 방법을 보여주며, 관련 시나리오의 컨텍스트에 맞는 작업을 볼 수 있습니다.
*시나리오*는 동일한 서비스 내에서 또는 다른 AWS 서비스와 결합된 상태에서 여러 함수를 직접적으로 호출하여 특정 태스크를 수행하는 방법을 보여주는 코드 예제입니다.
각 예시에는 전체 소스 코드에 대한 링크가 포함되어 있으며, 여기에서 컨텍스트에 맞춰 코드를 설정하고 실행하는 방법에 대한 지침을 찾을 수 있습니다.
**Topics**
+ [작업](#actions)
+ [시나리오](#scenarios)
+ [서버리스 예제](#serverless_examples)
## 작업
### `CreateQueue`
다음 코드 예시는 `CreateQueue`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
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
```
```
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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [CreateQueue](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/CreateQueue)를 참조하세요.
### `DeleteMessage`
다음 코드 예시는 `DeleteMessage`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [DeleteMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteMessage)를 참조하세요.
### `DeleteMessageBatch`
다음 코드 예시는 `DeleteMessageBatch`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
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
```
```
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 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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [DeleteMessageBatch](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteMessageBatch)를 참조하세요.
### `DeleteQueue`
다음 코드 예시는 `DeleteQueue`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
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
```
```
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 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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [DeleteQueue](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteQueue)를 참조하세요.
### `GetQueueAttributes`
다음 코드 예시는 `GetQueueAttributes`의 사용 방법을 보여 줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/topics_and_queues#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
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
```
+ API 세부 정보는 SDK for Python (Boto3) API 참조의 [GetQueueAttributes](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/GetQueueAttributes)를 참조하세요. *AWS *
### `GetQueueUrl`
다음 코드 예시는 `GetQueueUrl`의 사용 방법을 보여 줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [GetQueueUrl](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/GetQueueUrl)를 참조하세요.
### `ListQueues`
다음 코드 예시는 `ListQueues`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [ListQueues](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/ListQueues)를 참조하세요.
### `ReceiveMessage`
다음 코드 예시는 `ReceiveMessage`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
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
```
```
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 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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [ReceiveMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/ReceiveMessage)를 참조하세요.
### `SendMessage`
다음 코드 예시는 `SendMessage`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의[SendMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/SendMessage)를 참조하세요.
### `SendMessageBatch`
다음 코드 예시는 `SendMessageBatch`의 사용 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
```
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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 [SendMessageBatch](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/SendMessageBatch)를 참조하세요.
### `SetQueueAttributes`
다음 코드 예시는 `SetQueueAttributes`의 사용 방법을 보여 줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/topics_and_queues#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
주제에 대한 대기열의 정책 속성을 설정합니다.
```
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
```
+ API 세부 정보는 SDK for Python (Boto3) API 참조의 [SetQueueAttributes](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/SetQueueAttributes)를 참조하세요. *AWS *
## 시나리오
### 메신저 애플리케이션 생성
다음 코드 예제에서는 데이터베이스 테이블에서 메시지 레코드를 검색하는 AWS Step Functions 메신저 애플리케이션을 생성하는 방법을 보여줍니다.
**SDK for Python(Boto3)**
AWS SDK for Python (Boto3) 와 함께를 사용하여 Amazon DynamoDB 테이블에서 메시지 레코드를 검색하고 Amazon Simple Queue Service(Amazon SQS)를 통해 보내는 메신저 애플리케이션을 AWS Step Functions 생성하는 방법을 보여줍니다. 상태 시스템은 AWS Lambda 함수와 통합되어 데이터베이스에 전송되지 않은 메시지가 있는지 스캔합니다.
+ Amazon DynamoDB 테이블에서 메시지 레코드를 검색하고 업데이트하는 상태 머신을 생성합니다.
+ 상태 머신 정의를 업데이트하여 메시지를 Amazon Simple Queue Service(Amazon SQS)에도 전송합니다.
+ 상태 머신의 실행을 시작하고 중지합니다.
+ 서비스 통합을 사용하여 상태 머신에서 Lambda, DynamoDB 및 Amazon SQS에 연결합니다.
전체 소스 코드와 설정 및 실행 방법에 대한 지침은 [GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/stepfunctions_messenger)에서 전체 예제를 참조하세요.
**이 예제에서 사용되는 서비스**
+ DynamoDB
+ Lambda
+ Amazon SQS
+ 단계 함수
### Amazon Textract 탐색기 애플리케이션 생성
다음 코드 예제에서는 대화형 애플리케이션을 통해 Amazon Textract 출력을 탐색하는 방법을 보여줍니다.
**SDK for Python(Boto3)**
Amazon Textract와 AWS SDK for Python (Boto3) 함께를 사용하여 문서 이미지에서 텍스트, 양식 및 테이블 요소를 감지하는 방법을 보여줍니다. 입력 이미지와 Amazon Textract 출력은 탐지된 요소를 탐색할 수 있는 Tkinter 애플리케이션에 표시됩니다.
+ 문서 이미지를 Amazon Textract에 제출하고 감지된 요소의 출력을 탐색합니다.
+ Amazon Textract로 직접, 또는 Amazon Simple Storage Service(Amazon S3) 버킷을 통해 이미지를 제출합니다.
+ 비동기식 API를 사용하여 작업이 완료되면 Amazon Simple Notification Service(Amazon SNS) 주제에 알림을 게시하는 작업을 시작합니다.
+ Amazon Simple Queue Service(Amazon SQS) 대기열에서 작업 완료 메시지를 폴링하고 결과를 표시합니다.
전체 소스 코드와 설정 및 실행 방법에 대한 지침은 [GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/textract_explorer)에서 전체 예제를 참조하세요.
**이 예제에서 사용되는 서비스**
+ Amazon Cognito 자격 증명
+ Amazon S3
+ Amazon SNS
+ Amazon SQS
+ Amazon Textract
### FIFO 주제에 생성 및 게시
다음 코드 예제에서는 FIFO Amazon SNS 주제를 생성하고 거기에 게시하는 방법을 보여줍니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sns#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
Amazon FIFO 주제를 생성하고, Amazon SQS FIFO 및 표준 대기열에서 주제를 구독하고, 해당 주제에 메시지를 게시합니다.
```
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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 다음 주제를 참조하세요.
+ [CreateTopic](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/CreateTopic)
+ [게시](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Publish)
+ [Subscribe](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Subscribe)
### 동영상에서 사람과 객체 감지
다음 코드 예제에서는 Amazon Rekognition을 사용하여 동영상에서 사람과 객체를 감지하는 방법을 보여줍니다.
**SDK for Python(Boto3)**
Amazon Rekognition을 사용하여 비동기식 감지 작업을 시작해 동영상의 얼굴, 객체 및 사람을 감지할 수 있습니다. 또한 이 예제에서는 작업이 완료되고 주제에 대한 Amazon Simple Queue Service(Amazon SQS) 대기열을 구독할 때 Amazon Simple Notification Service(Amazon SNS) 주제를 알리도록 Amazon Rekognition을 구성합니다. 대기열이 작업에 대한 메시지를 받으면 작업이 검색되고 결과가 출력됩니다.
이 예제는 GitHub에서 가장 잘 볼 수 있습니다. 전체 소스 코드와 설정 및 실행 방법에 대한 지침은 [GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/rekognition)에서 전체 예제를 참조하세요.
**이 예제에서 사용되는 서비스**
+ Amazon Rekognition
+ Amazon S3
+ Amazon SES
+ Amazon SNS
+ Amazon SQS
### 대기열에 메시지 게시
다음 코드 예제에서는 다음과 같은 작업을 수행하는 방법을 보여줍니다.
+ 주제(FIFO 또는 비 FIFO)를 생성합니다.
+ 필터 적용 옵션을 사용하여 여러 개의 대기열로 주제를 구독합니다.
+ 주제에 메시지를 게시합니다.
+ 대기열에서 받은 메시지를 폴링합니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/cross_service/topics_and_queues#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
명령 프롬프트에서 대화형 시나리오를 실행합니다.
```
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.")
```
시나리오에서 사용할 Amazon SNS 및 Amazon SQS 작업을 래핑하는 클래스를 생성합니다.
```
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
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 다음 주제를 참조하세요.
+ [CreateQueue](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/CreateQueue)
+ [CreateTopic](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/CreateTopic)
+ [DeleteMessageBatch](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteMessageBatch)
+ [DeleteQueue](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteQueue)
+ [DeleteTopic](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/DeleteTopic)
+ [GetQueueAttributes](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/GetQueueAttributes)
+ [Publish](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Publish)
+ [ReceiveMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/ReceiveMessage)
+ [SetQueueAttributes](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/SetQueueAttributes)
+ [Subscribe](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Subscribe)
+ [Unsubscribe](https://docs.aws.amazon.com/goto/boto3/sns-2010-03-31/Unsubscribe)
### 메시지 일괄 전송 및 수신
다음 코드 예제에서는 다음과 같은 작업을 수행하는 방법을 보여줍니다.
+ Amazon SQS 대기열을 생성합니다.
+ 대기열에 메시지를 일괄 전송합니다.
+ 대기열에서 메시지를 일괄 수신합니다.
+ 대기열에서 메시지 배치를 삭제합니다.
**SDK for Python (Boto3)**
GitHub에 더 많은 내용이 있습니다. [AWS 코드 예 리포지토리](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/sqs#code-examples)에서 전체 예를 찾고 설정 및 실행하는 방법을 배워보세요.
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 response
```
래퍼 함수를 사용하여 메시지를 일괄적으로 보내고 받을 수 있습니다.
```
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)
```
+ API 세부 정보는 *AWS SDK for Python (Boto3) API 참조*의 다음 주제를 참조하세요.
+ [CreateQueue](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/CreateQueue)
+ [DeleteMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteMessage)
+ [DeleteMessageBatch](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteMessageBatch)
+ [DeleteQueue](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/DeleteQueue)
+ [ReceiveMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/ReceiveMessage)
+ [SendMessage](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/SendMessage)
+ [SendMessageBatch](https://docs.aws.amazon.com/goto/boto3/sqs-2012-11-05/SendMessageBatch)
## 서버리스 예제
### Amazon SQS 트리거에서 간접적으로 Lambda 함수 간접 호출
다음 코드 예제는 SQS 대기열에서 메시지를 받아 트리거된 이벤트를 수신하는 Lambda 함수를 구현하는 방법을 보여줍니다. 함수는 이벤트 파라미터에서 메시지를 검색하고 각 메시지의 내용을 로깅합니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [서버리스 예제](https://github.com/aws-samples/serverless-snippets/tree/main/integration-sqs-to-lambda) 리포지토리에서 전체 예제를 찾아보고 설정 및 실행 방법을 알아봅니다.
Python을 사용하여 Lambda로 SQS 이벤트를 사용합니다.
```
# 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
```
### Amazon SQS 트리거로 Lambda 함수에 대한 배치 항목 실패 보고
다음 코드 예제는 SQS 대기열에서 이벤트를 수신하는 Lambda 함수에 대한 부분 배치 응답을 구현하는 방법을 보여줍니다. 이 함수는 응답으로 배치 항목 실패를 보고하고 나중에 해당 메시지를 다시 시도하도록 Lambda에 신호를 보냅니다.
**SDK for Python(Boto3)**
GitHub에 더 많은 내용이 있습니다. [서버리스 예제](https://github.com/aws-samples/serverless-snippets/tree/main/lambda-function-sqs-report-batch-item-failures) 리포지토리에서 전체 예제를 찾아보고 설정 및 실행 방법을 알아봅니다.
Python을 사용하여 Lambda로 SQS 배치 항목 실패 보고
```
# 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
```