|
5 | 5 |
|
6 | 6 | import boto3 |
7 | 7 | from asyncer import asyncify |
8 | | -from mypy_boto3_sqs.service_resource import Queue |
| 8 | +from mypy_boto3_sqs.service_resource import Queue, SQSClient |
9 | 9 | from taskiq import AsyncBroker |
10 | 10 | from taskiq.abc.result_backend import AsyncResultBackend |
11 | 11 | from taskiq.acks import AckableMessage |
@@ -36,7 +36,7 @@ def __init__( |
36 | 36 | raise BrokerError("A valid SQS Queue URL is required") |
37 | 37 |
|
38 | 38 | self.sqs_queue_url = sqs_queue_url |
39 | | - self._sqs = boto3.resource("sqs") |
| 39 | + self._sqs: SQSClient = boto3.resource("sqs") |
40 | 40 | self._sqs_queue: Optional[Queue] = None |
41 | 41 |
|
42 | 42 | if max_number_of_messages > 10: |
@@ -75,17 +75,22 @@ async def kick( |
75 | 75 | # Must be explicitly set as a label to a unix timestamp |
76 | 76 | expiry = message.labels.pop("sqs_expiry", 0) |
77 | 77 |
|
78 | | - await asyncify(queue.send_message)( |
79 | | - # SQS structured message attributes |
80 | | - MessageAttributes={ |
81 | | - "expiry": { |
82 | | - "StringValue": str(expiry), |
83 | | - "DataType": "Number", |
84 | | - } |
85 | | - }, |
86 | | - MessageBody=message.message.decode("utf-8"), |
87 | | - MessageGroupId=message.task_name, |
88 | | - ) |
| 78 | + try: |
| 79 | + await asyncify(queue.send_message)( |
| 80 | + # SQS structured message attributes |
| 81 | + MessageAttributes={ |
| 82 | + "expiry": { |
| 83 | + "StringValue": str(expiry), |
| 84 | + "DataType": "Number", |
| 85 | + } |
| 86 | + }, |
| 87 | + MessageBody=message.message.decode("utf-8"), |
| 88 | + MessageGroupId=message.task_name, |
| 89 | + ) |
| 90 | + except Exception as err: |
| 91 | + # taskiq supresses the original exception, but it wold be good to know about |
| 92 | + logger.exception("Unhandled exception in SQSBroker") |
| 93 | + raise err |
89 | 94 |
|
90 | 95 | async def listen(self) -> AsyncGenerator[Union[bytes, AckableMessage], None]: |
91 | 96 | """ |
|
0 commit comments