kafkaesk

About the Project
Publish
Subscribe
Avoiding global object
Manual commit
kafkaesk contract
Worker
Development
Extensions
Naming

About The Project

This project is meant to help facilitate effortless publishing and subscribing to events with Python and Kafka.

Guiding principal

HTTP
Language agnostic
Contracts built on top of Kafka

Alternatives

aiokafka: can be complex to scale correctly
guillotina_kafka: complex, tied to Guillotina
faust: requires additional data layers, not language agnostic
confluent kafka + avro: close but ends up being like grpc. compilation for languages. No asyncio.

Consider this Python project as syntactic sugar around these ideas.

Publish

Using pydantic but can be done with pure JSON.

import kafkaesk
from pydantic import BaseModel

app = kafkaesk.Application()

@app.schema("Content", version=1, retention=24 * 60 * 60)
class ContentMessage(BaseModel):
    foo: str


async def foobar():
    # ...
    # doing something in an async func
    await app.publish("content.edited.Resource", data=ContentMessage(foo="bar"))

A convenience method is available in the subscriber dependency instance, this allow to header propagation from the consumed message.

import kafkaesk
from pydantic import BaseModel

app = kafkaesk.Application()

@app.schema("Content", version=1, retention=24 * 60 * 60)
class ContentMessage(BaseModel):
    foo: str


@app.subscribe("content.*", "group_id")
async def get_messages(data: ContentMessage, subscriber):
    print(f"{data.foo}")
    # This will propagate `data` record headers
    await subscriber.publish("content.edited.Resource", data=ContentMessage(foo="bar"))

import kafkaesk
from pydantic import BaseModel

app = kafkaesk.Application()

@app.schema("Content", version=1, retention=24 * 60 * 60)
class ContentMessage(BaseModel):
    foo: str


@app.subscribe("content.*", "group_id")
async def get_messages(data: ContentMessage):
    print(f"{data.foo}")

Avoiding global object

If you do not want to have global application configuration, you can lazily configure the application and register schemas/subscribers separately.

import kafkaesk
from pydantic import BaseModel

router = kafkaesk.Router()

@router.schema("Content", version=1, retention=24 * 60 * 60)
class ContentMessage(BaseModel):
    foo: str


@router.subscribe("content.*", "group_id")
async def get_messages(data: ContentMessage):
    print(f"{data.foo}")


if __name__ == "__main__":
    app = kafkaesk.Application()
    app.mount(router)
    kafkaesk.run(app)

Optional consumer injected parameters:

schema: str
record: aiokafka.structs.ConsumerRecord
app: kafkaesk.app.Application
subscriber: kafkaesk.app.BatchConsumer

Depending on the type annotation for the first parameter, you will get different data injected:

async def get_messages(data: ContentMessage): parses pydantic schema
async def get_messages(data: bytes): give raw byte data
async def get_messages(record: aiokafka.structs.ConsumerRecord): give kafka record object
async def get_messages(data): raw json data in message

Manual commit

To accomplish a manual commit strategy yourself:

app = kafkaesk.Application(auto_commit=False)

@app.subscribe("content.*", "group_id")
async def get_messages(data: ContentMessage, subscriber):
    print(f"{data.foo}")
    await subscriber.consumer.commit()

SSL

Add these values to your kafka_settings:

ssl_context - this should be a placeholder as the SSL Context is generally created within the application
security_protocol - one of SSL or PLAINTEXT
sasl_mechanism - one of PLAIN, GSSAPI, SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER
sasl_plain_username .
sasl_plain_password .

kafkaesk contract

This is a library around using kafka. Kafka itself does not enforce these concepts.

Every message must provide a json schema
Messages produced will be validated against json schema
Each topic will have only one schema
A single schema can be used for multiple topics
Consumed message schema validation is up to the consumer
Messages will be consumed at least once. Considering this, your handling should be idempotent

Message format

{
    "schema": "schema_name:1",
    "data": { ... }
}

Worker

kafkaesk mymodule:app --kafka-servers=localhost:9092

Options:

--kafka-servers: comma separated list of kafka servers
--kafka-settings: json encoded options to be passed to https://aiokafka.readthedocs.io/en/stable/api.html#aiokafkaconsumer-class
--topic-prefix: prefix to use for topics
--replication-factor: what replication factor topics should be created with. Defaults to min(number of servers, 3).

Application.publish

stream_id: str: name of stream to send data to
data: class that inherits from pydantic.BaseModel
key: Optional[bytes]: key for message if it needs one

Application.subscribe

stream_id: str: fnmatch pattern of streams to subscribe to
group: Optional[str]: consumer group id to use. Will use name of function if not provided

Application.schema

id: str: id of the schema to store
version: Optional[int]: version of schema to store
streams: Optional[List[str]]: if streams are known ahead of time, you can pre-create them before you push data
retention: Optional[int]: retention policy in seconds

Application.configure

kafka_servers: Optional[List[str]]: kafka servers to connect to
topic_prefix: Optional[str]: topic name prefix to subscribe to
kafka_settings: Optional[Dict[str, Any]]: additional aiokafka settings to pass in
replication_factor: Optional[int]: what replication factor topics should be created with. Defaults to min(number of servers, 3).
kafka_api_version: str: default auto
auto_commit: bool: default True
auto_commit_interval_ms: int: default 5000

Development

Requirements

poetry install

Run tests:

docker-compose up
KAFKA=localhost:9092 poetry run pytest tests

Extensions

Logging

This extension includes classes to extend Python's logging framework to publish structured log messages to a Kafka topic. This extension is made up of three main components: an extended logging.LogRecord and some custom logging.Handlers.

See logger.py in examples directory.

Log Record

kafkaesk.ext.logging.record.factory is a function that will return kafkaesk.ext.logging.record.PydanticLogRecord objects. The factory() function scans through any args passed to a logger and checks each item to determine if it is a subclass of pydantid.BaseModel.

If it is a base model instance and model._is_log_model evaluates to True the model will be removed from args and added to record._pydantic_data. After that factory() will use logging's existing logic to finish creating the log record.

Handler

This extensions ships with two handlers capable of handling kafkaesk.ext.logging.handler.PydanticLogModel classes: kafakesk.ext.logging.handler.PydanticStreamHandler and kafkaesk.ext.logging.handler.PydanticKafkaeskHandler.

The stream handler is a very small wrapper around logging.StreamHandler, the signature is the same, the only difference is that the handler will attempt to convert any pydantic models it receives to a human readable log message.

The kafkaesk handler has a few more bits going on in the background.

The handler has two required inputs, a kafkaesk.app.Application instance and a stream name.

Once initialized any logs emitted by the handler will be saved into an internal queue. There is a worker task that handles pulling logs from the queue and writing those logs to the specified topic.

Naming

It's hard and "kafka" is already a fun name. Hopefully this library isn't literally "kafkaesque" for you.

Release files for kafkaesk 0.8.7

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for kafkaesk 0.8.7
File	Size	Uploaded
kafkaesk-0.8.7.tar.gz	22.7 kB	Mar 19, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for kafkaesk 0.8.7
File	Interpreter	ABI	Platform	Reset
kafkaesk-0.8.7-py3-none-any.whl 23.7 kB Mar 19, 2026	Python 3	none	any	Details

Total release size: 46.4 kB

Release files / kafkaesk-0.8.7.tar.gz

Download URL	kafkaesk-0.8.7.tar.gz
Size	22.7 kB
Tags	Source
SHA-256 checksum How to use checksums	`3bf690d0a33bb2af9d028794aaedc112e2cf1a688e52cc1f961d70ae344668d6`
BLAKE2b-256 checksum How to use checksums	`82d478a817cdafda6bfbe8557f27c1cd802b29d15789d7244e621b882a330c5e`
Upload date	Mar 19, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`poetry/1.8.5 CPython/3.8.18 Linux/6.14.0-1017-azure`

Release files / kafkaesk-0.8.7-py3-none-any.whl

Download URL	kafkaesk-0.8.7-py3-none-any.whl
Size	23.7 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`173e6a697bc362bafadb373471d3db6650994adafd226607ea0f40792b55e275`
BLAKE2b-256 checksum How to use checksums	`9792f9267ba0ea34b6da2f0385d1f5cb778000e086071240ea8e6342f9e727c5`
Upload date	Mar 19, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`poetry/1.8.5 CPython/3.8.18 Linux/6.14.0-1017-azure`