Initial focus on embedded systems Broqer can be used wherever continuous streams of data have to be processed - and they are everywhere. Watch out!
Synopsis
Pure python implementation without dependencies
Under MIT license (2018 Günther Jena)
Source is hosted on GitHub.com
Documentation is hosted on ReadTheDocs.com
Tested on Python 3.10, 3.11, 3.12, 3.13 and 3.14
Unit tested with pytest, coding style checked with ruff, static type checked with mypy, static code checked with Pylint, documented with Sphinx
Operators known from ReactiveX and other streaming frameworks (like Map, CombineLatest, …)
Centralised object to keep track of publishers and subscribers
Starting point to build applications with a microservice architecture
Showcase
In other frameworks a Publisher is sometimes called Oberservable. A Subscriber is able to observe changes the publisher is emitting. With these basics you’re able to use the observer pattern - let’s see!
Observer pattern
Subscribing to a publisher is done via the .subscribe() method. A simple subscriber is Sink which is calling a function with optional positional and keyword arguments.
>>> from broqer import Publisher, Sink
>>> a = Publisher(5) # create a publisher with state `5`
>>> s = Sink(print, 'Change:') # create a subscriber
>>> disposable = a.subscribe(s) # subscribe subscriber to publisher
Change: 5
>>> a.notify(3) # change the state
Change: 3
>>> disposable.dispose() # unsubscribeCombine publishers with arithmetic operators
You’re able to create publishers on the fly by combining two publishers with the common operators (like +, >, <<, …).
>>> a = Publisher(1)
>>> b = Publisher(3)
>>> c = a * 3 > b # create a new publisher via operator overloading
>>> disposable = c.subscribe(Sink(print, 'c:'))
c: False
>>> a.notify(2)
c: True
>>> b.notify(10)
c: FalseAlso fancy stuff like getting item by index or key is possible:
>>> i = Publisher('a')
>>> d = Publisher({'a':100, 'b':200, 'c':300})
>>> disposable = d[i].subscribe(Sink(print, 'r:'))
r: 100
>>> i.notify('c')
r: 300
>>> d.notify({'c':123})
r: 123Some python built in functions can’t return Publishers (e.g. len() needs to return an integer). For these cases special functions are defined in broqer: Str, Int, Float, Len and In (for x in y). Also other functions for convenience are available: All, Any, BitwiseAnd and BitwiseOr.
Attribute access on a publisher is building a publisher where the actual attribute access is done on emitting values. A publisher has to know, which type it should mimic - this is done via .inherit_type(type).
>>> i = Publisher('Attribute access made REACTIVE')
>>> i.inherit_type(str)
>>> disposable = i.lower().split(sep=' ').subscribe(Sink(print))
['attribute', 'access', 'made', 'reactive']
>>> i.notify('Reactive and pythonic')
['reactive', 'and', 'pythonic']Function decorators
Make your own operators on the fly with function decorators. Decorators are available for Accumulate, CombineLatest, Filter, Map, MapAsync, MapThreaded, Reduce and Sink.
>>> from broqer import op
>>> @op.build_map
... def count_vowels(s):
... return sum([s.count(v) for v in 'aeiou'])
>>> msg = Publisher('Hello World!')
>>> disposable = (msg | count_vowels).subscribe(Sink(print, 'Number of vowels:'))
Number of vowels: 3
>>> msg.notify('Wahuuu')
Number of vowels: 4You can even make configurable Map s and Filter s:
>>> import re
>>> @op.build_filter_factory
... def filter_pattern(pattern, s):
... return re.search(pattern, s) is not None
>>> msg = Publisher('Cars passed: 135!')
>>> disposable = (msg | filter_pattern('[0-9]+')).subscribe(Sink(print))
Cars passed: 135!
>>> msg.notify('No cars have passed')
>>> msg.notify('Only 1 car has passed')
Only 1 car has passedInstall
pip install broqerCredits
Broqer was inspired by:
RxPY: Reactive Extension for Python (by Børge Lanes and Dag Brattli)
aioreactive: Async/Await reactive tools for Python (by Dag Brattli)
streamz: build pipelines to manage continuous streams of data (by Matthew Rocklin)
MQTT: M2M connectivity protocol
Florian Feurstein: spending hours of discussion, coming up with great ideas and help me understand the concepts!
API
Publishers
A Publisher is the source for messages.
Publisher () |
Basic publisher |
Operators
CombineLatest (*publishers) |
Combine the latest emit of multiple publishers and emit the combination |
Filter (predicate, …) |
Filters values based on a predicate function |
Map (map_func, *args, **kwargs) |
Apply map_func(*args, value, **kwargs) to each emitted value |
MapAsync (coro, mode, …) |
Apply coro(*args, value, **kwargs) to each emitted value |
Throttle (duration) |
Limit the number of emits per duration |
Subscribers
A Subscriber is the sink for messages.
Sink (func, *args, **kwargs) |
Apply func(*args, value, **kwargs) to each emitted value |
SinkAsync (coro, …) |
Apply coro(*args, value, **kwargs) to each emitted value |
OnEmitFuture (timeout=None) |
Build a future able to await for |
Trace (d) |
Debug output for publishers |
Values
Value (*init) |
Publisher and Subscriber |
1 maintainer
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file broqer-3.3.0.tar.gz.
File metadata
- Download URL: broqer-3.3.0.tar.gz
- Upload date:
- Size: 60.6 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.1.0 CPython/3.13.7
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
fb949927b247ecde0195d59213fe7fb16762b8ecc3b6b38b028208da3c8d8c26
|
|
| MD5 |
b5ae3a3a05517c48f0d21ccca6a504fb
|
|
| BLAKE2b-256 |
6ba20286f299bca6c9a537217a08aed7a10cc79b5214d7b7aa4803e13d41e490
|
File details
Details for the file broqer-3.3.0-py2.py3-none-any.whl.
File metadata
- Download URL: broqer-3.3.0-py2.py3-none-any.whl
- Upload date:
- Size: 35.4 kB
- Tags: Python 2, Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.1.0 CPython/3.13.7
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
2b74bb37999aca8637480f31aed928f35e737746a3f7257767d33e07c6797b35
|
|
| MD5 |
bd4e14161236df03a662cf185f03fbc8
|
|
| BLAKE2b-256 |
7bd87bf4e09dd7c9064d0c4e2eff024acec9a96dbfb2f9720a6d5d9a95089d9b
|
