taskiq-sqs

This library provides SQS broker and S3 result backend for TaskIQ.

Installation

pip install taskiq-sqs

Basic usage

Here is an example of how to use the SQS broker with the S3 backend:

importasynciofromtaskiq_sqsimportS3Bucket, S3ResultBackend, SQSBrokerQUEUE_NAME="my-queue"broker=SQSBroker(
"http://localhost:4566/000000000000/my-queue", # specify existing queuesqs_region_override="us-east-1"
).with_result_backend(
S3ResultBackend(
bucket=S3Bucket(name="response-bucket") # by default backend will create bucket for you if it does not exist
)
)
@broker.task()asyncdefi_love_aws() ->None:
awaitasyncio.sleep(1)
print("Hello there!")
asyncdefmain() ->None:
awaitbroker.startup()
task=awaiti_love_aws.kiq()
print(awaittask.wait_result())
awaitbroker.shutdown()
if__name__=="__main__":
asyncio.run(main())

How to run:

• run worker first with taskiq worker examples.example_broker:broker
• after that run broker to create a task and wait for result: python examples/example_broker.py

Message expiration

If you set the sqs_expiry label to a unix timestamp, the message will be discarded if the worker receives it after that time.

importasynciofromtaskiq_sqsimportSQSBrokerbroker=SQSBroker("http://sqs.us-east-1.localhost.localstack.cloud:4566/000000000000/my-queue")
@broker.taskasyncdefadd_one(value: int) ->int:
returnvalue+1asyncdefmain() ->None:
# Never forget to call startup in the beginning.awaitbroker.startup()
# Send the task to the broker.task=awaitadd_one.kiq(1)
# Wait for the result. (result backend must be configured)result=awaittask.wait_result(timeout=2)
print(f"Task execution took: {result.execution_time} seconds.")
ifnotresult.is_err:
print(f"Returned value: {result.return_value}")
else:
print("Error found while executing task.")
awaitbroker.shutdown()
if__name__=="__main__":
asyncio.run(main())