A simple backend agnostic heartbeating convention
Project description
beatit
A simple backend agnostic heartbeating convention.
>>> class Printer:
... @staticmethod
... def publish(*, subject: bytes, payload: bytes):
... print(f"{payload} -> {subject}")
>>> from beatit import Heart
>>> heart = Heart(process="my.process.identifier", publisher=Printer)
>>> heart.start(warmup=60)
b'start/60' -> b'heartbeat.my.process.identifier'
>>> heart.beat(period=5)
b'beat/5' -> b'heartbeat.my.process.identifier'
>>> heart.beat(period=5)
>>> heart.degraded(period=5)
b'degraded/5' -> b'heartbeat.my.process.identifier'
>>> heart.degraded(period=5)
>>> heart.stop()
b'stop' -> b'heartbeat.my.process.identifier'
All you need is a publisher with a publish
method accepting two named arguments subject
and payload
.
Instantiate a Heart
instance with a process
name and a publisher
. beats will be published on the subject heartbeat.<process>
.
What are the beats?
start
with a warmup
period (after which a beat
or a degraded
is expected)
beat
with a period
(when to expect the next beat
).
degraded
with a period
(when to expect a beat
or a degraded
next).
stop
when the process stops gracefully.
sync or not ...
If you favour async (what is wrong with you?) Heart recognizes an async publisher automagically and all you have to do is await all the things.
>>> import asyncio
>>> class AsyncPrinter:
... @staticmethod
... async def publish(*, subject: bytes, payload: bytes):
... print(f"{payload} -> {subject}")
>>> from beatit import Heart
>>> heart = Heart(process="my.process.identifier", publisher=AsyncPrinter)
>>> asyncio.run(heart.start(warmup=60))
b'start/60' -> b'heartbeat.my.process.identifier'
>>> asyncio.run(heart.beat(period=5))
b'beat/5' -> b'heartbeat.my.process.identifier'
>>> asyncio.run(heart.beat(period=5))
>>> asyncio.run(heart.degraded(period=5))
b'degraded/5' -> b'heartbeat.my.process.identifier'
>>> asyncio.run(heart.degraded(period=5))
>>> asyncio.run(heart.stop())
b'stop' -> b'heartbeat.my.process.identifier'
subject_as_string
beatit
calls publish
with bytes
subject and payload. Seemed consistent but common publishers favour str
for subject.
Fear not, here is a @limx0 approved solution!
>>> class StrBytes:
... @staticmethod
... def publish(*, subject: str, payload: bytes):
... print(f"{payload} -> {subject}")
>>> from beatit import Heart
>>> heart = Heart(process="my.process.identifier", publisher=StrBytes, subject_as_string=True)
>>> heart.start(warmup=30)
b'start/30' -> heartbeat.my.process.identifier
Project details
Release history Release notifications | RSS feed
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
File details
Details for the file beatit-2022.3.1.tar.gz
.
File metadata
- Download URL: beatit-2022.3.1.tar.gz
- Upload date:
- Size: 4.0 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: python-requests/2.27.1
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | aa6660f6edf60a150f8c4f05077f5680c4cc2daaf1f179088571880b644c0578 |
|
MD5 | 51590d03c14eced7ad2799eea76cc88b |
|
BLAKE2b-256 | 2e6fb097b187097ebc972db7c86d467635f624371bb94c09764de13bb110d17e |
File details
Details for the file beatit-2022.3.1-py2.py3-none-any.whl
.
File metadata
- Download URL: beatit-2022.3.1-py2.py3-none-any.whl
- Upload date:
- Size: 3.5 kB
- Tags: Python 2, Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: python-requests/2.27.1
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 76cf511c2d2a190785ef746d6fbc0f1855534145fee9527138412bb1551d74af |
|
MD5 | 239193b79505d751ab2c62de9658b899 |
|
BLAKE2b-256 | 3579c765f15a1ae50642d19a8a16d3b4994b6dac19ff841d818ffb72f00ffd17 |