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.1.1.tar.gz (8.9 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.1.1-py3-none-any.whl (11.4 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: muffin_kafka-3.1.1.tar.gz
  • Upload date:
  • Size: 8.9 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.12.0 {"installer":{"name":"uv","version":"0.12.0","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.1.1.tar.gz
Algorithm Hash digest
SHA256 17839b9770cf078b48866a1773670c7a03e94d08a100a6d8e398206290b32a54
MD5 ec4e38f7dec82455d88025432c62a424
BLAKE2b-256 c3846493f93c13bb330538cb62656e0b060052823e2755079941c50f7139a5e7

See more details on using hashes here.

File details

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

File metadata

  • Download URL: muffin_kafka-3.1.1-py3-none-any.whl
  • Upload date:
  • Size: 11.4 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: uv/0.12.0 {"installer":{"name":"uv","version":"0.12.0","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.1.1-py3-none-any.whl
Algorithm Hash digest
SHA256 d9f8201b037a03623b499feaa0f1c66197a5bead5c4bc6cd26b4d7d93891fcab
MD5 0f49362c531994f71dfa1372b99b2d72
BLAKE2b-256 ccd4d8ce489dd56cebcc78eb1eaa8467797af8a24e83be823273ed73f20a825c

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