brq
brq is a lightweight python library for job queue based on the redis stream, with no central server and self-organized by Consumer.
Prerequisites
Redis >= 6.2, tested with latest redis 6/7 docker image. Recommended to use redis>=7, which includes more inspection features.
Install
pip install brq
Feature
See examples for running examples.
- Defer job and automatic retry error job
- Dead queue for unprocessable job, you can process it later
- Multiple consumers in one consumer group
- No scheduler needed, consumer handles itself
- Using callback function to process job result or exception
Configuration
If using BrqConfig(for example, @task), you can use a .env file and environment variables to configure brq. The prefix of environment variables is BRQ_.
For example,
BRQ_REDIS_PORT=6379 python consumer.pyfor specifying redis port.
See configs for more details.
Echo job overview
Producer
import os
from brq.producer import Producer
from brq.configs import BrqConfig
async def main():
config = BrqConfig()
async with config.open_redis_client() as async_redis_client:
await Producer(
async_redis_client,
redis_prefix=config.redis_key_prefix,
redis_seperator=config.redis_key_seperator,
max_message_len=config.producer_max_message_length,
).run_job("echo", ["hello"])
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Consumer
The only thing you need is @task, and the target function can be sync or async and sync function will be converted to async function and run in a thread automatically.
from brq import task
@task
def echo(message):
print(f"Received message: {message}")
if __name__ == "__main__":
# Run the task once, for local debug
# echo("hello")
# Run as a daemon
echo.serve()
This is the same as the following, the classic way...But more flexible.
import os
from brq.consumer import Consumer
from brq.daemon import Daemon
from brq.tools import get_redis_client, get_redis_url
async def echo(message):
print(message)
async def main():
redis_url = get_redis_url(
host=os.getenv("REDIS_HOST", "localhost"),
port=int(os.getenv("REDIS_PORT", 6379)),
db=int(os.getenv("REDIS_DB", 0)),
cluster=bool(os.getenv("REDIS_CLUSTER", "false") in ["True", "true", "1"]),
tls=bool(os.getenv("REDIS_TLS", "false") in ["True", "true", "1"]),
username=os.getenv("REDIS_USERNAME", ""),
password=os.getenv("REDIS_PASSWORD", ""),
)
async with get_redis_client(redis_url) as async_redis_client:
daemon = Daemon(Consumer(async_redis_client, echo))
await daemon.run_forever()
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Technical details: deferred jobs
We can use defer_until as a datetime or defer_hours+defer_minutes+defer_seconds to calculate a timestamp based on current redis timestamp. And use unique to set the job to be unique or not.
By default, unique=True means Job with the exactly same function_name, args and kwargs will be unique, which allows the same Job to add into the deferred queue more than once. In this case, we differentiate tasks by the current redis timestamp(Job.create_at) and an additional uuid(Job.uid), just like redis stream did.
If unique=False, the same Job will be added into the deferred queue only once. Duplicates will update the job's defer time. In this case, you can use your own uuid in args(or kwargs) to differentiate Job.
Develop
Install pre-commit before commit
pip install pre-commit
pre-commit install
Install package locally
pip install -e .[test]
Run unit-test before PR
pytest -v
1 maintainer
Metadata
1 maintainer
Release files for brq 0.5.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| brq-0.5.0.tar.gz | 80.2 kB | Without provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| brq-0.5.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 101.7 kB
Release files / brq-0.5.0.tar.gz
| Download URL | brq-0.5.0.tar.gz |
|---|---|
| Size | 80.2 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
1f4ac67654ab43e90a1d9e37fa7eedad0b207811751b83a6e9c843dc24090b51
|
|
BLAKE2b-256 checksum How to use checksums |
503b57cd088c62cb6b685e7f3ab14adde550daab02380c0d7bbbe7ad3403115e
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/5.1.1 CPython/3.12.7
|
Release files / brq-0.5.0-py3-none-any.whl
| Download URL | brq-0.5.0-py3-none-any.whl |
|---|---|
| Size | 21.4 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
ad2f8789768bca3b78a0b88db04d89cd2fd6ff68ce22916be9e4d5e6d1af79ff
|
|
BLAKE2b-256 checksum How to use checksums |
170deb7a26ba499638c6e4be6ce8517946a349c7ac10113fdd14114572f850bf
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/5.1.1 CPython/3.12.7
|
Security for brq 0.5.0
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
