<div align="center">
<img src="img/pyinsole-img.png" style="width:100px;height:100px;" alt="Circular Image">
</div>
<br>
<p align="center">
<em><b>pyinsole</b> is an asynchronous message dispatcher inpired by <a href="https://github.com/georgeyk/loafer">loafer</a> designed to provide a flexible and efficient way to consume messages from Amazon SQS queues. The <b>pyinsole</b> simplifies the process of integrating with SQS by offering multiple consumption strategies, allowing you to choose the best approach for your application's needs.</em>
</p>
<br>
## 💻 Usage
The script defines an asynchronous message handler function (`my_handler`) that will be invoked whenever a message is received from the SQS queue. The `SQSRoute` class is used to route messages from the `example-queue` to the handler.
#### Example Code
Here’s the main code that processes messages from the `example-queue`. The script will listen for messages on the `example-queue`, and for each message received, it will print the message content, associated metadata, and any additional keyword arguments.
```python
import os
from pyinsole import Manager
from pyinsole.ext.aws import SQSRoute
async def my_handler(message: dict, metadata: dict, **kwargs):
print(f"message={message}, metadata={metadata}, kwargs={kwargs}")
return True
provider_options = {
"endpoint_url": os.getenv("AWS_ENDPOINT_URL"),
"options": {
"MaxNumberOfMessages": 10,
"WaitTimeSeconds": os.getenv("AWS_WAIT_TIME_SECONDS", 20),
},
}
routes = [
SQSRoute('example-queue', handler=my_handler, provider_options=provider_options),
]
if __name__ == '__main__':
manager = Manager(routes)
manager.run()
```
Or you can use class based handlers if you prefer:
```python
import os
from pyinsole import Manager
from pyinsole.ext.aws import SQSRoute
from pyinsole.ext.handlers import AsyncHandler
class MyHandler(AsyncHandler):
async def process(self, message, metadata, **kwargs) -> bool:
print(f"message={message}, metadata={metadata}, kwargs={kwargs}")
return True
provider_options = {
"endpoint_url": os.getenv("AWS_ENDPOINT_URL"),
"options": {
"MaxNumberOfMessages": 10,
"WaitTimeSeconds": os.getenv("AWS_WAIT_TIME_SECONDS", 20),
},
}
routes = [
SQSRoute('example-queue', handler=MyHandler(), provider_options=provider_options),
]
if __name__ == '__main__':
manager = Manager(routes)
manager.run()
```
#### Running the Script
This setup allows you to easily process messages from an SQS queue using the `pyinsole` library. You can modify the `my_handler` function to implement your specific message processing logic.
1. **Start LocalStack** (or ensure you have access to AWS SQS). If you are using LocalStack, make sure it's running and the `example-queue` is created:
```bash
aws --endpoint-url=http://localhost:4566 sqs create-queue --queue-name example-queue
```
2. **Run the script**:
```bash
python your_script.py
```
3. **Push some messages**:
```
aws --endpoint-url=http://localhost:4566 sqs send-message --queue-url http://localhost:4566/000000000000/example-queue --message-body "Your message body"
```
<br>
## 🎯 Roadmap
You can find the project roadmap [here](./ROADMAP.md).
This document outlines future improvements, features, and tasks planned for the project.
<br>
## 🫱🏻🫲🏽 How to contribute
We welcome contributions of all kinds to make **pyinsole** better! To contribute to the project, follow these steps:
1. **Fork the repository**: Click on the "Fork" button at the top right of the repository page.
2. **Clone your fork**:
```bash
git clone https://github.com/edopneto/pyinsole
cd pyinsole
```
3. **Create a new branch**: It's best practice to create a feature branch for your changes.
```bash
git checkout -b feature/your-feature-name
```
4. **Make your changes**: Work on your feature, bug fix, or documentation improvement.
5. **Test your changes**: Ensure everything is working as expected.
6. **Commit your changes**:
```bash
git add .
git commit -m "Add a brief message describing your changes"
```
7. **Push to your branch**:
```bash
git push origin feature/your-feature-name
```
8. **Open a Pull Request**: Go to the repository on GitHub, and you’ll see a button to "Compare & Pull Request." Submit a pull request with a clear title and description of your changes.
Raw data
{
"_id": null,
"home_page": null,
"name": "pyinsole",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "asyncio, dispatcher, message, microservices, tasks",
"author": null,
"author_email": "Eduardo Neto <edopneto@gmail.com>, Christian Hartung <hartung@live.com>",
"download_url": "https://files.pythonhosted.org/packages/25/2e/c4a47e45d566d2d934df1c3ddc25e89f8b349c90afcbf6e30e3d240bdeae/pyinsole-0.2.0a3.tar.gz",
"platform": null,
"description": "<div align=\"center\">\n <img src=\"img/pyinsole-img.png\" style=\"width:100px;height:100px;\" alt=\"Circular Image\">\n</div>\n<br>\n<p align=\"center\">\n <em><b>pyinsole</b> is an asynchronous message dispatcher inpired by <a href=\"https://github.com/georgeyk/loafer\">loafer</a> designed to provide a flexible and efficient way to consume messages from Amazon SQS queues. The <b>pyinsole</b> simplifies the process of integrating with SQS by offering multiple consumption strategies, allowing you to choose the best approach for your application's needs.</em>\n</p>\n<br>\n\n## \ud83d\udcbb Usage\n\nThe script defines an asynchronous message handler function (`my_handler`) that will be invoked whenever a message is received from the SQS queue. The `SQSRoute` class is used to route messages from the `example-queue` to the handler.\n\n#### Example Code\n\nHere\u2019s the main code that processes messages from the `example-queue`. The script will listen for messages on the `example-queue`, and for each message received, it will print the message content, associated metadata, and any additional keyword arguments.\n\n```python\nimport os\n\nfrom pyinsole import Manager\nfrom pyinsole.ext.aws import SQSRoute\n\nasync def my_handler(message: dict, metadata: dict, **kwargs):\n print(f\"message={message}, metadata={metadata}, kwargs={kwargs}\")\n return True\n\nprovider_options = {\n \"endpoint_url\": os.getenv(\"AWS_ENDPOINT_URL\"),\n \"options\": {\n \"MaxNumberOfMessages\": 10,\n \"WaitTimeSeconds\": os.getenv(\"AWS_WAIT_TIME_SECONDS\", 20),\n },\n}\n\nroutes = [\n SQSRoute('example-queue', handler=my_handler, provider_options=provider_options),\n]\n\nif __name__ == '__main__':\n manager = Manager(routes)\n manager.run()\n```\n\nOr you can use class based handlers if you prefer:\n\n```python\nimport os\n\nfrom pyinsole import Manager\nfrom pyinsole.ext.aws import SQSRoute\nfrom pyinsole.ext.handlers import AsyncHandler\n\nclass MyHandler(AsyncHandler):\n async def process(self, message, metadata, **kwargs) -> bool:\n print(f\"message={message}, metadata={metadata}, kwargs={kwargs}\")\n return True\n\nprovider_options = {\n \"endpoint_url\": os.getenv(\"AWS_ENDPOINT_URL\"),\n \"options\": {\n \"MaxNumberOfMessages\": 10,\n \"WaitTimeSeconds\": os.getenv(\"AWS_WAIT_TIME_SECONDS\", 20),\n },\n}\n\nroutes = [\n SQSRoute('example-queue', handler=MyHandler(), provider_options=provider_options),\n]\n\nif __name__ == '__main__':\n manager = Manager(routes)\n manager.run()\n```\n\n\n#### Running the Script\n\nThis setup allows you to easily process messages from an SQS queue using the `pyinsole` library. You can modify the `my_handler` function to implement your specific message processing logic.\n\n1. **Start LocalStack** (or ensure you have access to AWS SQS). If you are using LocalStack, make sure it's running and the `example-queue` is created:\n ```bash\n aws --endpoint-url=http://localhost:4566 sqs create-queue --queue-name example-queue\n ```\n\n2. **Run the script**:\n ```bash\n python your_script.py\n ```\n\n3. **Push some messages**:\n ```\n aws --endpoint-url=http://localhost:4566 sqs send-message --queue-url http://localhost:4566/000000000000/example-queue --message-body \"Your message body\"\n ```\n\n<br>\n\n## \ud83c\udfaf Roadmap\n\nYou can find the project roadmap [here](./ROADMAP.md).\n\nThis document outlines future improvements, features, and tasks planned for the project.\n\n<br>\n\n## \ud83e\udef1\ud83c\udffb\u200d\ud83e\udef2\ud83c\udffd How to contribute\n\nWe welcome contributions of all kinds to make **pyinsole** better! To contribute to the project, follow these steps:\n\n1. **Fork the repository**: Click on the \"Fork\" button at the top right of the repository page.\n\n2. **Clone your fork**:\n ```bash\n git clone https://github.com/edopneto/pyinsole\n cd pyinsole\n ```\n\n3. **Create a new branch**: It's best practice to create a feature branch for your changes.\n ```bash\n git checkout -b feature/your-feature-name\n ```\n\n4. **Make your changes**: Work on your feature, bug fix, or documentation improvement.\n\n5. **Test your changes**: Ensure everything is working as expected.\n\n6. **Commit your changes**:\n ```bash\n git add .\n git commit -m \"Add a brief message describing your changes\"\n ```\n\n7. **Push to your branch**:\n ```bash\n git push origin feature/your-feature-name\n ```\n\n8. **Open a Pull Request**: Go to the repository on GitHub, and you\u2019ll see a button to \"Compare & Pull Request.\" Submit a pull request with a clear title and description of your changes.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Asynchronous message dispatcher for concurrent tasks processing",
"version": "0.2.0a3",
"project_urls": {
"Documentation": "https://github.com/pyinsole/pyinsole#readme",
"Issues": "https://github.com/pyinsole/pyinsole/issues",
"Source": "https://github.com/pyinsole/pyinsole"
},
"split_keywords": [
"asyncio",
" dispatcher",
" message",
" microservices",
" tasks"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "4200d19286fb7e34e744cd86910740f2924b7fb11054dc16f4a169e7a02903a1",
"md5": "41ca41bc5f3154a5e806b2b6d9dc9795",
"sha256": "94b2ab743f537c4e928a3e9b3222f19772d6d33fb4effdd5d2bdd0b8ba24cb9d"
},
"downloads": -1,
"filename": "pyinsole-0.2.0a3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "41ca41bc5f3154a5e806b2b6d9dc9795",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 15206,
"upload_time": "2024-12-10T15:28:48",
"upload_time_iso_8601": "2024-12-10T15:28:48.198380Z",
"url": "https://files.pythonhosted.org/packages/42/00/d19286fb7e34e744cd86910740f2924b7fb11054dc16f4a169e7a02903a1/pyinsole-0.2.0a3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "252ec4a47e45d566d2d934df1c3ddc25e89f8b349c90afcbf6e30e3d240bdeae",
"md5": "2f802f60cb0586226339dc789022b16e",
"sha256": "e0448f6c9286745af6c9f9876a643d91a3b74641858e08934a84157142632699"
},
"downloads": -1,
"filename": "pyinsole-0.2.0a3.tar.gz",
"has_sig": false,
"md5_digest": "2f802f60cb0586226339dc789022b16e",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 706637,
"upload_time": "2024-12-10T15:28:49",
"upload_time_iso_8601": "2024-12-10T15:28:49.894517Z",
"url": "https://files.pythonhosted.org/packages/25/2e/c4a47e45d566d2d934df1c3ddc25e89f8b349c90afcbf6e30e3d240bdeae/pyinsole-0.2.0a3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-10 15:28:49",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "pyinsole",
"github_project": "pyinsole#readme",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "pyinsole"
}
JSON
