Robust, high-volume, message based communication made easy.
kiwiPy is a library that makes remote messaging using RabbitMQ (and possibly other message brokers) EASY. It was designed to support high-throughput workflows in big-data and computational science settings and is currently used by AiiDA for computational materials research around the world. That said, kiwiPy is entirely general and can be used anywhere where high-throughput and robust messaging are needed.
Here’s what you get:
Broadcast (with filters)
Task queue messages
Let’s dive in, with some examples taken from the rmq tutorial. To see more detail head over to the documentation.
import kiwipy with kiwipy.connect('amqp://localhost') as comm: # Send an RPC message print(" [x] Requesting fib(30)") response = comm.rpc_send('fib', 30).result() print((" [.] Got %r" % response))
import threading import kiwipy def fib(comm, num): if num == 0: return 0 if num == 1: return 1 return fib(comm, num - 1) + fib(comm, num - 2) with kiwipy.connect('amqp://127.0.0.1') as comm: # Register an RPC subscriber with the name 'fib' comm.add_rpc_subscriber(fib, 'fib') # Now wait indefinitely for fibonacci calls threading.Event().wait()
Create a new task:
import sys import kiwipy message = ' '.join(sys.argv[1:]) or "Hello World!" with rmq.connect('amqp://localhost') as comm: comm.task_send(message)
And the worker:
import time import threading import kiwipy print(' [*] Waiting for messages. To exit press CTRL+C') def callback(_comm, task): print((" [x] Received %r" % task)) time.sleep(task.count(b'.')) print(" [x] Done") try: with kiwipy.connect('amqp://localhost') as comm: comm.add_task_subscriber(callback) threading.Event().wait() except KeyboardInterrupt: pass
If you use kiwiPy directly or indirectly (e.g. by using AiiDA) then please cite:
Uhrin, M., & Huber, S. P. (2020). kiwiPy : Robust , high-volume , messaging for big-data and computational science workflows, 5, 4–6. http://doi.org/10.21105/joss.02351
This helps us to keep making community software.
This software follows Semantic Versioning
Want a new feature? Found a bug? Want to contribute more documentation or a translation perhaps?
Help is always welcome, get started with the contributing guide.
This package utilises tox for unit test automation, and pre-commit for code style formatting and test automation.
To install these development dependencies:
pip install tox pre-commit
To run the unit tests:
For the rmq tests you will require a running instance of RabbitMQ. One way to achieve this is using Docker and launching test/rmq/docker-compose.yml.
To run the pre-commit tests:
pre-commit run --all
To build the documentation:
tox -e docs-clean
Changes should be submitted as Pull Requests (PRs) to the develop branch.
Create a release PR/commit to the develop branch, updating kiwipy/version.py and CHANGELOG.md.
Fast-forward merge develop into the master branch
Create a release on GitHub (https://github.com/aiidateam/kiwipy/releases/new), pointing to the release commit on master, named v.X.Y.Z (identical to version in kiwipy/version.py)
This will trigger the continuous-deployment GitHub workflow which, if all tests pass, will publish the package to PyPi. Check this has successfully completed in the GitHub Actions tab (https://github.com/aiidateam/kiwipy/actions).
(if the release fails, delete the release and tag)
Release history Release notifications | RSS feed
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.