Muffin-Kafka

Muffin-Kafka is an Apache Kafka integration plugin for the Muffin web framework, built on top of aiokafka.

Tests Status PYPI Version Python Versions

Features

  • Async Kafka integration using aiokafka
  • Single or batch message consumption — stream messages one-by-one or read in batches
  • Per-topic task model — each topic is consumed in an isolated asyncio task
  • Simple handler registration using @plugin.handle_topics(...)
  • Manual or auto-commit support, custom group IDs
  • Producer support via send
  • Built-in monitoring with offsets, lag, and poll delay
  • Healthcheck manage command for liveness probes and observability
  • Optional error handler via @plugin.handle_error(...)

Requirements

  • Python ≥ 3.11
  • Muffin ≥ 0.71
  • Kafka cluster or broker (local or cloud)

Installation

pip install muffin-kafka

Usage

    from muffin import Application
    from muffin_kafka import Kafka

    app = Application("example")

    # Initialize the plugin with config options
    kafka = Kafka(app, bootstrap_servers="localhost:9092", produce=True, listen=True)

Registering Handlers

Use @kafka.handle_topics(...) to register a handler for specific Kafka topics:

    @kafka.handle_topics("events.user", "events.auth")
    async def handle_event(message):
        data = message.value.decode()
        print("Received:", data)

You can also register a global error handler:

    @kafka.handle_error
    async def on_error(exc):
        print("Kafka error:", exc)

Sending Messages

You can send messages to Kafka topics using the send method:

    # Send a message
    await kafka.send("events.user", {"action": "signup"}, key="user123")

Listening (kafka-listen)

Start consuming messages using the kafka-listen management command:

    # Listen to all registered handlers
    muffin myapp kafka-listen

    # Listen to specific topics only
    muffin myapp kafka-listen events.user events.order

    # Override group id for this run
    muffin myapp kafka-listen --group-id=workers-v2

    # Enable monitoring with custom interval
    muffin myapp kafka-listen --monitor --monitor-interval=30

    # Batch mode (process messages in batches)
    muffin myapp kafka-listen --batch-size=100

CLI Options:

Option Description
topics Specific topics to listen (optional, defaults to all registered handlers)
--group-id Override consumer group ID
--monitor Enable built-in monitoring
--monitor-interval Monitoring interval in seconds (default: 60)
--batch-size Read messages in batches (uses getmany())

Healthcheck

Run the kafka-healthcheck management command to check consumer lag:

    muffin myapp kafka-healthcheck

Or programmatically via Muffin's manage command:

    # Check specific topics
    await app.manage.commands["kafka-healthcheck"]("events.user", max_lag=1000)

Monitoring

If monitor=True is passed, the plugin will log:

  • Committed offsets
  • Latest offsets
  • Poll timestamps
  • Per-partition lag and delay

This data can be extended for Prometheus/Grafana metrics or alerting.

Configuration Options

You can pass configuration options either as keyword arguments to the plugin:

kafka = Kafka(app, bootstrap_servers="localhost:9092", produce=True)

Or set them via Muffin's config system (e.g. .env, YAML):

"KAFKA_BOOTSTRAP_SERVERS": "localhost:9092",
"KAFKA_PRODUCE": True,

Supported Options

Option Type Default Description
bootstrap_servers str "localhost:9092" Kafka broker connection string
group_id str None Kafka consumer group ID
client_id str "muffin" Kafka client ID
produce bool False Enable Kafka producer
listen bool True Enable consumers (message listening)
monitor bool False Enable internal consumer monitor
batch_size int None Read messages in batches (uses getmany())
setup_commands bool True Register management commands (kafka-healthcheck, kafka-listen)
monitor_interval int 60 Monitor frequency in seconds
auto_offset_reset str "earliest" Where to start if no committed offset
enable_auto_commit bool False Auto-commit offsets after each message/batch. When False, the plugin commits manually after all handlers succeed.
max_poll_records int None Max records to poll in one batch
request_timeout_ms int 30000 Request timeout
retry_backoff_ms int 1000 Retry interval on failure
security_protocol str "PLAINTEXT" Kafka protocol (SSL, SASL_PLAINTEXT, …)
sasl_mechanism str "PLAIN" SASL auth mechanism
sasl_plain_username str None SASL auth user
sasl_plain_password str None SASL auth password
ssl_cafile str None Path to trusted CA certs

Bug Tracker

Found a bug or have a feature request? Please open an issue at: https://github.com/klen/muffin-kafka/issues

Contributing

Pull requests are welcome! Development happens here: https://github.com/klen/muffin-kafka

License

MIT – See LICENSE for full details.

Project links

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Newer release available (3.1.4)

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for klen from gravatar.com klen

Credits

PyPI data
Data sourced directly from PyPI's database.

Author: Kirill Klenov

License expression

MIT
View SPDX License List

Requires

Python <4, >=3.11

Tags

kafka asyncio asgi muffin

Classifiers

Report project as malware

Download files

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

Source Distribution

muffin_kafka-3.0.0.tar.gz (8.6 kB view details)

Uploaded Source

Built Distribution

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

muffin_kafka-3.0.0-py3-none-any.whl (11.3 kB view details)

Uploaded Python 3

File details

Details for the file muffin_kafka-3.0.0.tar.gz.

File metadata

  • Download URL: muffin_kafka-3.0.0.tar.gz
  • Upload date:
  • Size: 8.6 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.11.24 {"installer":{"name":"uv","version":"0.11.24","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}

File hashes

Hashes for muffin_kafka-3.0.0.tar.gz
Algorithm Hash digest
SHA256 00b14b99dba514d855f40614174e9d16aa63e4cd8e0b3b5af43cf1358e4aedcb
MD5 490ce18a99be97e3a14ddbd87cd6ff4d
BLAKE2b-256 8a5dc6ba26e9e3d2f78eb95937f031899b7b60ff8a0f401e2f59e467c0862e25

See more details on using hashes here.

File details

Details for the file muffin_kafka-3.0.0-py3-none-any.whl.

File metadata

  • Download URL: muffin_kafka-3.0.0-py3-none-any.whl
  • Upload date:
  • Size: 11.3 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.11.24 {"installer":{"name":"uv","version":"0.11.24","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}

File hashes

Hashes for muffin_kafka-3.0.0-py3-none-any.whl
Algorithm Hash digest
SHA256 9dd5127f40bcd57e6c790c8c7247a0fcd8bfb4f69c89887833288b1c7c5a51dc
MD5 475a837cd2553a4a3680de99c937686b
BLAKE2b-256 ca0e40a58db13a19efb454d13fd37fc2292fc5df3305e954fb29afb4fa512071

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page