eventiq

Asyncio native pub/sub framework for Python

Tests Build License Mypy Python Format PyPi

Installation

pip install eventiq

poetry add eventiq

Installing optional dependencies

pip install 'eventiq[broker]'

Available brokers

nats
rabbitmq
kafka
redis

Features

Modern, asyncio based python 3.8+ syntax
Fully type annotated
Minimal external dependencies (anyio, pydantic, typer)
Automatic message parsing based on type annotations using pydantic
Code hot-reload
Highly scalable: each service can process hundreds of tasks concurrently, all messages are load balanced between all instances by default
Resilient - at least once delivery for all messages by default (except for Redis*)
Customizable & pluggable message encoder/decoder (json as default)
Multiple broker support
- Memory (for testing)
- Nats
- Kafka
- Rabbitmq
- Redis
Result Backend implementation for Nats & Redis
Lifespan protocol support
Lightweight (and completely optional) dependency injection system based on type annotations
Easy and lightweight (~3k lines of code including types definitions and brokers implementations)
Cloud Events standard as base message structure (no more python specific *args and **kwargs in messages)
AsyncAPI documentation generation from code
Twelve factor app approach - stdout logging, configuration through environment variables
Easily extensible via Middlewares
Multiple extensions and integrations including:
- Prometheus - mertics exporter
- OpenTelemetry - tracing and metrics
- Message Pack - message pack encoder for messages
- FastAPI - integrating eventiq Service with FastAPI applications (WIP)
- Dataref - data reference resolver for messages (WIP)
- Eventiq Workflows - orchestration engine built on top of eventiq (WIP)

Basic Usage

import asyncio
from eventiq import Service, Middleware, CloudEvent, GenericConsumer
from eventiq.backends.nats import JetStreamBroker

class SendMessageMiddleware(Middleware):
    async def after_broker_connect(self):
        print(f"After service start, running with {service.broker}")
        await asyncio.sleep(10)
        for i in range(100):
            message = CloudEvent(topic="test.topic", data={"counter": i})
            await service.publish(message)
        print("Published messages(s)")

broker = JetStreamBroker(url="nats://localhost:4222")

service = Service(
    name="example-service",
    broker=broker,
)
service.add_middleware(SendMessageMiddleware)

@service.subscribe(topic="test.topic")
async def example_run(message: CloudEvent):
    print(f"Received Message {message.id} with data: {message.data}")

@service.subscribe(topic="test.topic2")
class MyConsumer(GenericConsumer[CloudEvent]):
    async def process(self, message: CloudEvent):
        print(f"Received Message {message.id} with data: {message.data}")
        await self.publish(CloudEvent(topic="test.topic", data={"response": "ok"})

Run with

eventiq run app:service --log-level=info

Watching for changes

eventiq run app:service --log-level=info --reload=.

Testing

StubBroker class is provided as in memory replacement for running unit tests

import os


def get_broker(**kwargs):
    if os.getenv('ENV') == 'TEST':
        from eventiq.backends.stub import StubBroker
        return StubBroker()
    else:
        from eventiq.backends.rabbitmq import RabbitmqBroker
        return RabbitmqBroker(**kwargs)

broker = get_broker()

Furthermore, subscribers are just regular python coroutines, so it's possible to test them simply by invocation

# main.py
@service.subscribe(topic="test.topic")
async def my_subscriber(message: CloudEvent):
    return 42

# tests.py
from main import my_subscriber

async def test_my_subscriber():
    result = await my_subscriber(None)
    assert result == 42

CLI

Getting help:

eventiq --help

Installing shell autocompletion:

eventiq --install-completion [bash|zsh|fish|powershell|pwsh]

Basic commands

run - run service
docs - generate AsyncAPI docs
send - send message to broker

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

eventiq-1.1.16.tar.gz (38.7 kB view details)

Uploaded Aug 19, 2026 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

eventiq-1.1.16-py3-none-any.whl (45.9 kB view details)

Uploaded Aug 19, 2026 Python 3

File details

Details for the file eventiq-1.1.16.tar.gz.

File metadata

Download URL: eventiq-1.1.16.tar.gz
Upload date: Aug 19, 2026
Size: 38.7 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/7.0.0 CPython/3.13.14

File hashes

Hashes for eventiq-1.1.16.tar.gz
Algorithm	Hash digest
SHA256	`95a0bf8818a65e8a3657ed723d2bbd4a40fb651e6392c34c342b83853c645118`
MD5	`2f179be297f636424f2b3d39adbafe09`
BLAKE2b-256	`704f328e44489e78bc65c9e7693bc00495529ea75bddf914a56874d7b1add2c5`

See more details on using hashes here.

File details

Details for the file eventiq-1.1.16-py3-none-any.whl.

File metadata

Download URL: eventiq-1.1.16-py3-none-any.whl
Upload date: Aug 19, 2026
Size: 45.9 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/7.0.0 CPython/3.13.14

File hashes

Hashes for eventiq-1.1.16-py3-none-any.whl
Algorithm	Hash digest
SHA256	`0e6f592ba10aac4cc250fa5893507c267423074db36e189757f1b92107f7641f`
MD5	`a9b7a89f1f385a4f8a776eb2f6858f29`
BLAKE2b-256	`2ed688410e2f914bfc5f9eef02c23ee0e69b62f38ff39031d090f4a3c763c4e8`