================
amqp_helper
================
Introduction
=============
:code:`amqp_helper` aims to be a simple Helper library to configure AMQP communication for use with :code:`aio-pika`
To achieve this goal this Library provides the :code:`AMQPConfig` class.
This package also provides a log handler to send logs to an AMQP Broker. You can use it via the class :code:`AMQPLogHandler`
Installation
==============
:code:`amqp_helper` can be installed in multiple ways. The easiest Solution is to install it with :code:`pip`.
via pip
---------
.. code-block:: bash
python3 -m pip install amqp-helper
from source
------------
.. code-block:: bash
git clone https://github.com/bad-microservices/amqp_helper.git
cd amqp_helper
python3 -m pip install .
Example
========
.. code-block:: python
import asyncio
from amqp_helper import AMQPConfig
from aio_pika import connect_robust
amqp_config = AMQPConfig(username="test",password="testpw",vhost="testvhost")
async def main():
connection = await connect_robust(**amqp_config.aio_pika())
# do some amqp stuff
if __name__ == "__main__":
asyncio.run(main())
Example RPC over AMQP
======================
Server code
------------
The Server code is quite simple
.. code-block:: python
import asyncio
from amqp_helper import AMQPConfig, AMQPService, new_amqp_func
amqp_config = AMQPConfig(username="test",password="testpw",vhost="testvhost")
async def testfunc(throw_value_error = False,throw_key_error = False, throw_exception = False*args, **kwargs):
if throw_value_error:
raise ValueError()
if throw_key_error:
raise KeyError()
if throw_exception:
raise Exception()
return {"result": "sync stuff"}
rpc_fun = new_amqp_func("test1", test1234)
@rpc_fun.exception_handler(ValueError, KeyError)
async def handle_value_error(*args, **kwargs):
retrun "got ValueError or KeyError"
@rpc_fun.exception_handler(Exception)
async def handle_value_error(*args, **kwargs):
return "got Exception"
async def main():
service = await AMQPService().connect(amqp_config)
await service.register_function(rpc_fun)
await service.serve()
# do some amqp stuff
if __name__ == "__main__":
asyncio.run(main())
Client
------------
.. code-block:: python
import asyncio
from amqp_helper import AMQPConfig, AMQPClient
amqp_config = AMQPConfig(username="test",password="testpw",vhost="testvhost")
async def main():
client = await AMQPClient().connect(amqp_config)
print(await client.call(None,"test1"))
if __name__ == "__main__":
asyncio.run(main())
Logging to AMQP
================
if we want to log to an AMQP Topic we can do it with the following example code.
.. code-block:: python
import logging
from amqp_helper import AMQPLogHandler
log_cfg = AMQPConfig(username="test",password="testpw",vhost="testvhost")
handler = AMQPLogHandler(amqp_config=log_cfg, exchange_name="amqp.topic")
root_logger= logging.getLogger()
root_logger.addHandler(handler)
root_logger.info("test log message")
Raw data
{
"_id": null,
"home_page": "https://github.com/bad-microservices/amqp_helper",
"name": "amqp-helper",
"maintainer": "",
"docs_url": null,
"requires_python": "",
"maintainer_email": "",
"keywords": "amqp",
"author": "Ole Hannemann",
"author_email": "cerberus885@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/d2/d8/ef00a7dd2a38a2fd3a11daa4467cc6d56f67aa3472fccb284b7197a5126c/amqp_helper-0.1.4.tar.gz",
"platform": null,
"description": "================\namqp_helper\n================\n\nIntroduction\n=============\n\n:code:`amqp_helper` aims to be a simple Helper library to configure AMQP communication for use with :code:`aio-pika`\nTo achieve this goal this Library provides the :code:`AMQPConfig` class.\n\nThis package also provides a log handler to send logs to an AMQP Broker. You can use it via the class :code:`AMQPLogHandler`\n\nInstallation\n==============\n\n:code:`amqp_helper` can be installed in multiple ways. The easiest Solution is to install it with :code:`pip`.\n\nvia pip\n---------\n\n.. code-block:: bash\n\n python3 -m pip install amqp-helper\n\n\nfrom source\n------------\n\n.. code-block:: bash\n\n git clone https://github.com/bad-microservices/amqp_helper.git\n cd amqp_helper\n python3 -m pip install .\n\nExample\n========\n\n.. code-block:: python\n\n import asyncio\n from amqp_helper import AMQPConfig\n from aio_pika import connect_robust\n\n amqp_config = AMQPConfig(username=\"test\",password=\"testpw\",vhost=\"testvhost\")\n\n async def main():\n\n connection = await connect_robust(**amqp_config.aio_pika())\n\n # do some amqp stuff\n\n if __name__ == \"__main__\":\n asyncio.run(main())\n\nExample RPC over AMQP\n======================\n\nServer code\n------------\nThe Server code is quite simple\n\n.. code-block:: python\n\n import asyncio\n from amqp_helper import AMQPConfig, AMQPService, new_amqp_func\n\n amqp_config = AMQPConfig(username=\"test\",password=\"testpw\",vhost=\"testvhost\")\n\n async def testfunc(throw_value_error = False,throw_key_error = False, throw_exception = False*args, **kwargs):\n if throw_value_error:\n raise ValueError()\n if throw_key_error:\n raise KeyError()\n if throw_exception:\n raise Exception()\n\n return {\"result\": \"sync stuff\"}\n\n rpc_fun = new_amqp_func(\"test1\", test1234)\n\n\n @rpc_fun.exception_handler(ValueError, KeyError)\n async def handle_value_error(*args, **kwargs):\n retrun \"got ValueError or KeyError\"\n\n @rpc_fun.exception_handler(Exception)\n async def handle_value_error(*args, **kwargs):\n return \"got Exception\"\n\n async def main():\n\n service = await AMQPService().connect(amqp_config)\n await service.register_function(rpc_fun)\n\n await service.serve()\n\n # do some amqp stuff\n\n if __name__ == \"__main__\":\n asyncio.run(main())\n\n\nClient\n------------\n\n.. code-block:: python\n\n import asyncio\n from amqp_helper import AMQPConfig, AMQPClient\n\n amqp_config = AMQPConfig(username=\"test\",password=\"testpw\",vhost=\"testvhost\")\n\n async def main():\n\n client = await AMQPClient().connect(amqp_config)\n\n print(await client.call(None,\"test1\"))\n\n if __name__ == \"__main__\":\n asyncio.run(main())\n\n\nLogging to AMQP\n================\n\nif we want to log to an AMQP Topic we can do it with the following example code.\n\n.. code-block:: python\n\n import logging\n from amqp_helper import AMQPLogHandler\n\n log_cfg = AMQPConfig(username=\"test\",password=\"testpw\",vhost=\"testvhost\")\n\n handler = AMQPLogHandler(amqp_config=log_cfg, exchange_name=\"amqp.topic\")\n\n root_logger= logging.getLogger()\n root_logger.addHandler(handler)\n\n root_logger.info(\"test log message\")\n\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "simple Helper library to configure AMQP communication",
"version": "0.1.4",
"project_urls": {
"Homepage": "https://github.com/bad-microservices/amqp_helper"
},
"split_keywords": [
"amqp"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "a0d45e8dbada963af2fbfd94b5c14a52e107cc1e1d4429e803dd90c7b0cdaf7e",
"md5": "0e6eea67c9c579e71bca9c9bfb56c5cd",
"sha256": "9e133a6d42ef4b2454901e2cfc5885190d0d1abb5d42278de2740261136cbbd8"
},
"downloads": -1,
"filename": "amqp_helper-0.1.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0e6eea67c9c579e71bca9c9bfb56c5cd",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": null,
"size": 9488,
"upload_time": "2023-09-11T09:21:32",
"upload_time_iso_8601": "2023-09-11T09:21:32.348031Z",
"url": "https://files.pythonhosted.org/packages/a0/d4/5e8dbada963af2fbfd94b5c14a52e107cc1e1d4429e803dd90c7b0cdaf7e/amqp_helper-0.1.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "d2d8ef00a7dd2a38a2fd3a11daa4467cc6d56f67aa3472fccb284b7197a5126c",
"md5": "7c2a4c736159e2c2adfd6dfe95eec664",
"sha256": "56fb4eff5cd427f719ffafbfb9779573fabaed0d2d5218c1ab833ba33a998522"
},
"downloads": -1,
"filename": "amqp_helper-0.1.4.tar.gz",
"has_sig": false,
"md5_digest": "7c2a4c736159e2c2adfd6dfe95eec664",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 8480,
"upload_time": "2023-09-11T09:21:33",
"upload_time_iso_8601": "2023-09-11T09:21:33.467564Z",
"url": "https://files.pythonhosted.org/packages/d2/d8/ef00a7dd2a38a2fd3a11daa4467cc6d56f67aa3472fccb284b7197a5126c/amqp_helper-0.1.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-09-11 09:21:33",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "bad-microservices",
"github_project": "amqp_helper",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "amqp-helper"
}
JSON
