# Async SQS consumer
Python asynchronous (**async** / **await**) worker for consuming messages
from AWS SQS.
This is a hobby project, if you find the project interesting
any contribution is welcome.
## Usage
You must create an instance of the worker with the url of the queue.
Aws credentials are taken from environment variables, you must set the
following environment variables. Or you can provide a Context object with the
aws credentials `async_sqs_consumer.types.Context`
- `AWS_ACCESS_KEY_ID`
- `AWS_SECRET_ACCESS_KEY`
Example:
You can get the queue url with the follow aws cli command
`aws sqs get-queue-url --queue-name xxxxxx`
```python
# test_worker.py
from async_sqs_consumer.worker import (
Worker,
)
worker = Worker(
queue_url="https://sqs.us-east-1.amazonaws.com/xxxxxxx/queue_name"
)
@worker.task("report")
async def report(text: str) -> None:
print(text)
if __name__: "__main__":
worker.start()
```
Now you can initialize the worker `python test_worker.py`
Now you need to post a message for the worker to process
```python
import json
import boto3
import uuid
client = boto3.client("sqs")
client.send_message(
QueueUrl="https://sqs.us-east-1.amazonaws.com/xxxxxxx/queue_name",
MessageBody=json.dumps(
{
"task": "report",
"id": uuid.uuid4().hex,
"args": ["hello world"],
}
),
)
```
Or you can use aioboto3
```python
import asyncio
import json
import aioboto3
import uuid
async def main() -> None:
session = aioboto3.Session()
async with session.client("sqs") as client:
await client.send_message(
QueueUrl="https://sqs.us-east-1.amazonaws.com/xxxxxxx/queue_name",
MessageBody=json.dumps(
{
"task": "report",
"id": uuid.uuid4().hex,
"args": ["hello world"],
}
),
)
if __name__ == "__main__":
asyncio.run(main())
```
To publish the messages they must have the following structure
```json
{
"type": "object",
"properties": {
"task": {"type": "string"},
"id": {"type": "string"},
"args": {"type": "array"},
"kwargs": {"type": "object"},
"retries": {"type": "number"},
"eta": {"type": "string"},
"expires": {"type": "string"},
},
"required": ["task", "id"],
}
```
Raw data
{
"_id": null,
"home_page": "https://github.com/drestrepom/async_sqs_consumer",
"name": "async-sqs-consumer",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.9,<4.0",
"maintainer_email": "",
"keywords": "AWS,AWS,SQS,sqs,consumer,async,worker",
"author": "Diego Restrepo",
"author_email": "drestrepo@fluidattacks.com",
"download_url": "https://files.pythonhosted.org/packages/e3/15/c9825ec20f363c1520e2d8de8ccbaf5b80a055220f19b49bd9992039d608/async_sqs_consumer-0.3.8.tar.gz",
"platform": null,
"description": "# Async SQS consumer\n\nPython asynchronous (**async** / **await**) worker for consuming messages\nfrom AWS SQS.\n\nThis is a hobby project, if you find the project interesting\nany contribution is welcome.\n\n## Usage\n\nYou must create an instance of the worker with the url of the queue.\n\nAws credentials are taken from environment variables, you must set the\nfollowing environment variables. Or you can provide a Context object with the\naws credentials `async_sqs_consumer.types.Context`\n\n- `AWS_ACCESS_KEY_ID`\n- `AWS_SECRET_ACCESS_KEY`\n\nExample:\n\nYou can get the queue url with the follow aws cli command\n`aws sqs get-queue-url --queue-name xxxxxx`\n\n```python\n# test_worker.py\n\nfrom async_sqs_consumer.worker import (\n Worker,\n)\n\nworker = Worker(\n queue_url=\"https://sqs.us-east-1.amazonaws.com/xxxxxxx/queue_name\"\n)\n\n\n@worker.task(\"report\")\nasync def report(text: str) -> None:\n print(text)\n\nif __name__: \"__main__\":\n worker.start()\n```\n\nNow you can initialize the worker `python test_worker.py`\n\nNow you need to post a message for the worker to process\n\n```python\nimport json\nimport boto3\nimport uuid\n\nclient = boto3.client(\"sqs\")\n\nclient.send_message(\n QueueUrl=\"https://sqs.us-east-1.amazonaws.com/xxxxxxx/queue_name\",\n MessageBody=json.dumps(\n {\n \"task\": \"report\",\n \"id\": uuid.uuid4().hex,\n \"args\": [\"hello world\"],\n }\n ),\n)\n```\n\nOr you can use aioboto3\n\n```python\nimport asyncio\nimport json\nimport aioboto3\nimport uuid\n\n\nasync def main() -> None:\n session = aioboto3.Session()\n async with session.client(\"sqs\") as client:\n await client.send_message(\n QueueUrl=\"https://sqs.us-east-1.amazonaws.com/xxxxxxx/queue_name\",\n MessageBody=json.dumps(\n {\n \"task\": \"report\",\n \"id\": uuid.uuid4().hex,\n \"args\": [\"hello world\"],\n }\n ),\n )\n\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\nTo publish the messages they must have the following structure\n\n```json\n{\n \"type\": \"object\",\n \"properties\": {\n \"task\": {\"type\": \"string\"},\n \"id\": {\"type\": \"string\"},\n \"args\": {\"type\": \"array\"},\n \"kwargs\": {\"type\": \"object\"},\n \"retries\": {\"type\": \"number\"},\n \"eta\": {\"type\": \"string\"},\n \"expires\": {\"type\": \"string\"},\n },\n \"required\": [\"task\", \"id\"],\n}\n```\n",
"bugtrack_url": null,
"license": "",
"summary": "",
"version": "0.3.8",
"split_keywords": [
"aws",
"aws",
"sqs",
"sqs",
"consumer",
"async",
"worker"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "cf58dea2ac04251349f2e044ec90646a985b772d77731e77b102bf13998919be",
"md5": "3c4b573a4af5655135a25d06f0828a06",
"sha256": "7b96c90e27bdc2af34889053f5285b40733a59cea27d419171fede797990041b"
},
"downloads": -1,
"filename": "async_sqs_consumer-0.3.8-py3-none-any.whl",
"has_sig": false,
"md5_digest": "3c4b573a4af5655135a25d06f0828a06",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9,<4.0",
"size": 11430,
"upload_time": "2023-04-12T22:39:31",
"upload_time_iso_8601": "2023-04-12T22:39:31.156639Z",
"url": "https://files.pythonhosted.org/packages/cf/58/dea2ac04251349f2e044ec90646a985b772d77731e77b102bf13998919be/async_sqs_consumer-0.3.8-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "e315c9825ec20f363c1520e2d8de8ccbaf5b80a055220f19b49bd9992039d608",
"md5": "5f32cd53363fcff04d268ae936624aee",
"sha256": "1afdcafc0a494de6fec8027f4c12fc6dd805535cbdaf6330da1b831a244a45ed"
},
"downloads": -1,
"filename": "async_sqs_consumer-0.3.8.tar.gz",
"has_sig": false,
"md5_digest": "5f32cd53363fcff04d268ae936624aee",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9,<4.0",
"size": 9697,
"upload_time": "2023-04-12T22:39:34",
"upload_time_iso_8601": "2023-04-12T22:39:34.496865Z",
"url": "https://files.pythonhosted.org/packages/e3/15/c9825ec20f363c1520e2d8de8ccbaf5b80a055220f19b49bd9992039d608/async_sqs_consumer-0.3.8.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-04-12 22:39:34",
"github": true,
"gitlab": false,
"bitbucket": false,
"github_user": "drestrepom",
"github_project": "async_sqs_consumer",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "async-sqs-consumer"
}
JSON
