Taskiq NATS
Taskiq-nats is a plugin for taskiq that adds NATS broker. This package has support for NATS JetStream.
Installation
To use this project you must have installed core taskiq library:
pip install taskiq taskiq-nats
Usage
Here's a minimal setup example with a broker and one task.
Default NATS broker.
import asyncio
from taskiq_nats import NatsBroker, JetStreamBroker
broker = NatsBroker(
[
"nats://nats1:4222",
"nats://nats2:4222",
],
queue="random_queue_name",
)
@broker.task
async def my_lovely_task():
print("I love taskiq")
async def main():
await broker.startup()
await my_lovely_task.kiq()
await broker.shutdown()
if __name__ == "__main__":
asyncio.run(main())
NATS broker based on JetStream
import asyncio
from taskiq_nats import (
PushBasedJetStreamBroker,
PullBasedJetStreamBroker
)
broker = PushBasedJetStreamBroker(
servers=[
"nats://nats1:4222",
"nats://nats2:4222",
],
queue="awesome_queue_name",
)
# Or you can use pull based variant
broker = PullBasedJetStreamBroker(
servers=[
"nats://nats1:4222",
"nats://nats2:4222",
],
durable="awesome_durable_consumer_name",
)
@broker.task
async def my_lovely_task():
print("I love taskiq")
async def main():
await broker.startup()
await my_lovely_task.kiq()
await broker.shutdown()
if __name__ == "__main__":
asyncio.run(main())
NatsBroker configuration
Here's the constructor parameters:
-
servers
- a single string or a list of strings with nats nodes addresses. -
subject
- name of the subect that will be used to exchange tasks betwee workers and clients. -
queue
- optional name of the queue. By default NatsBroker broadcasts task to all workers, but if you want to handle every task only once, you need to supply this argument. -
result_backend
- custom result backend. -
task_id_generator
- custom function to generate task ids. - Every other keyword argument will be sent to
nats.connect
function.
JetStreamBroker configuration
Common
-
servers
- a single string or a list of strings with nats nodes addresses. -
subject
- name of the subect that will be used to exchange tasks betwee workers and clients. -
stream_name
- name of the stream where subjects will be located. -
queue
- a single string or a list of strings with nats nodes addresses. -
result_backend
- custom result backend. -
task_id_generator
- custom function to generate task ids. -
stream_config
- a config for stream. -
consumer_config
- a config for consumer.
PushBasedJetStreamBroker
-
queue
- name of the queue. It's used to share messages between different consumers.
PullBasedJetStreamBroker
-
durable
- durable name of the consumer. It's used to share messages between different consumers. -
pull_consume_batch
- maximum number of message that can be fetched each time. -
pull_consume_timeout
- timeout for messages fetch. If there is no messages, we start fetching messages again.