A command dispatcher and message bus implementation for python.
Home Page: https://banshee.artisan.io
A command dispatcher and message bus implementation for python.
poetry add banshee
See the Documentation or ask questions on the Discussion board.
This project is licensed under the MIT licence.
All documentation and images are licenced under the Creative Commons Attribution-ShareAlike 4.0 International License.
This project uses Semantic Versioning.
![dependabot[bot] avatar](https://avatars.githubusercontent.com/in/29110?v=4 "dependabot[bot]")
![github-actions[bot] avatar](https://avatars.githubusercontent.com/in/15368?v=4 "github-actions[bot]")
Handle messages by routing them to specific handlers.
It means that the commander and commandee can be separated without any real knowledge of each other.
A DispatchMiddleware that routes requests to handlers based on a preconfigured mapping.
A registry will need to provide a means to associate a request type with a handler.
Using the Injector Module multiple times in one process causes a crash, this prevents some system test configurations.
To be able to create and use multiple containers with the module.
=================================================== test session starts ===================================================
platform linux -- Python 3.11.6, pytest-7.4.2, pluggy-1.3.0 -- /VENV/bin/python
cachedir: .pytest_cache
rootdir: /ROOT
configfile: pyproject.toml
plugins: mypy-0.10.3, pylint-0.21.0, cov-4.1.0, pydocstyle-2.3.2, typeguard-4.1.5, anyio-3.7.1, asyncio-0.21.1
asyncio: mode=Mode.STRICT
collected 1 item
tests/bug.py::test_bug FAILED [100%]
======================================================== FAILURES =========================================================
________________________________________________________ test_bug _________________________________________________________
def test_bug() -> None:
r = banshee.Registry()
c = injector.Injector(banshee.extra.injector.BansheeModule(r))
c.get(banshee.Bus)
> c = injector.Injector(banshee.extra.injector.BansheeModule(r))
tests/bug.py:10:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/VENV/lib/python3.11/site-packages/injector/__init__.py:923: in __init__
self.binder.install(module)
/VENV/lib/python3.11/site-packages/injector/__init__.py:564: in install
instance(self)
/VENV/lib/python3.11/site-packages/injector/__init__.py:869: in __call__
self.configure(binder)
/VENV/lib/python3.11/site-packages/banshee/extra/injector.py:95: in configure
binder.bind(injector.inject(banshee.IdentityMiddleware))
/VENV/lib/python3.11/site-packages/injector/__init__.py:1410: in inject
inject(cast(Any, constructor_or_class).__init__)
/VENV/lib/python3.11/site-packages/injector/__init__.py:1415: in inject
read_and_store_bindings(function, bindings)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
f = <slot wrapper '__init__' of 'object' objects>, bindings = {}
@private
def read_and_store_bindings(f: Callable, bindings: Dict[str, type]) -> None:
function_bindings = getattr(f, '__bindings__', None) or {}
if function_bindings == 'deferred':
function_bindings = {}
merged_bindings = dict(function_bindings, **bindings)
if hasattr(f, '__func__'):
f = cast(Any, f).__func__
> cast(Any, f).__bindings__ = merged_bindings
E AttributeError: 'wrapper_descriptor' object has no attribute '__bindings__'
/VENV/lib/python3.11/site-packages/injector/__init__.py:1477: AttributeError
================================================= short test summary info =================================================
FAILED tests/bug.py::test_bug - AttributeError: 'wrapper_descriptor' object has no attribute '__bindings__'
==================================================== 1 failed in 0.06s ====================================================
The problem seems to stem from calling injector.inject() on a class after it has already been used for injection.
This happens in the configure method of BansheeModule.
The calls seem to be superfluous and can be removed without changing functionality.
import banshee
import banshee.extra.injector
import injector
def test_bug() -> None:
r = banshee.Registry()
c = injector.Injector(banshee.extra.injector.BansheeModule(r))
c.get(banshee.Bus)
c = injector.Injector(banshee.extra.injector.BansheeModule(r))Allow a message to be delegated to "transports" where it will be passed to a background worker.
It mean's work can be carried out by background workers, or sent to other micro services, rather than being dependent on the receiving process.
An abstract Transport class supporting the following operations:
read -- read and unserialise messages from the channelsend -- serialise and send a message to the channelack -- acknowledge that a message was processed successfullyreject -- prevent any retrying of a messageThis would be used by a SendMiddleware and Worker class which would handle adding messages to the transport and processing them.
The registry will need to provide a means to associate a request type with a transport.
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
Django
The Web framework for perfectionists with deadlines.
Laravel
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
Microsoft
Open source projects and samples from Microsoft.
Google
Google ❤️ Open Source for everyone.
Alibaba
Alibaba Open Source for everyone
D3
Data-Driven Documents codes.
Tencent
China tencent open source team.