kafka-component
Async Kafka producer/consumer components for python-components.
Install
uv add kafka-component
Requirements
- Python >= 3.11
- A running Kafka broker (KRaft or ZooKeeper mode)
Usage
import asyncio
from python_components import System
from kafka_component import KafkaProducerComponent, KafkaConsumerComponent, DeadLetterPolicy
producer = KafkaProducerComponent(bootstrap_servers="localhost:9092")
async def handle_order(value):
print("received order", value)
consumer = KafkaConsumerComponent(
bootstrap_servers="localhost:9092",
group_id="orders-service",
topics=["orders"],
handler=handle_order,
error_policy=DeadLetterPolicy(producer, dlq_topic="orders.DLQ"),
).using(["producer"])
system = System({"producer": producer, "consumer": consumer})
async def main():
async with system:
await producer.send("orders", {"order_id": 1})
asyncio.run(main())
The consumer's .using(["producer"]) declares a start-order dependency on the
producer so System's topological sort always starts the producer first —
it is not read as an injected self.producer attribute by
KafkaConsumerComponent; the DeadLetterPolicy above already holds a
direct reference to the producer instance.
Semantics and caveats
- At-least-once delivery. The consumer commits offsets only after the handler succeeds. A crash between a successful handler call and the commit can redeliver a message — handlers should be idempotent if that matters.
- Sequential processing. One message is handled at a time, in partition order. There is no built-in concurrency; scale via more consumer instances/partitions, not intra-instance parallelism.
- Not a supervisor. On handler failure, the configured
ErrorPolicyruns (defaultSkipAndLogPolicy, orDeadLetterPolicy) and the loop continues — the component itself does not retry, crash-loop, or restart. Health is exposed viaroutes(); process supervision is the caller's job. - JSON value-only. Message values are
json.dumps/json.loads. Keys and headers pass through untouched. - Graceful shutdown drains the in-flight message.
shutdown()waits for the currently-processing message's handler (and itsErrorPolicy, if it fails) to finish before stopping — it does not abandon work mid-message.
Development
uv sync --all-groups
uv run pytest
uv run ruff format --check .
uv run ruff check .
Integration tests spin up a real Kafka broker via testcontainers. For local development against a persistent broker instead, run docker compose up -d.
License
MIT
1 maintainer
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file kafka_component-0.1.0.tar.gz.
File metadata
- Download URL: kafka_component-0.1.0.tar.gz
- Upload date:
- Size: 17.4 kB
- Tags: Source
- Uploaded using Trusted Publishing? Yes
- Uploaded via:
twine/7.0.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
de6bc3d4496c52969668b1ce7405e50629a7ea4a54dd457668a4a940a1bbacf0
|
|
| MD5 |
37fd6a92580bcaab41aef7b61cf89935
|
|
| BLAKE2b-256 |
12d20556623f81fa4cb7d85f8c92f7c707059fc08ace33dc82fcb6ce279b2b7d
|
Provenance
The following attestation bundles were made for kafka_component-0.1.0.tar.gz:
Publisher:
publish-pypi.yml on Fabriciooml/kafka-component
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
kafka_component-0.1.0.tar.gz -
Subject digest:
de6bc3d4496c52969668b1ce7405e50629a7ea4a54dd457668a4a940a1bbacf0 - Sigstore transparency entry: 2743034305
- Sigstore integration time:
-
Permalink:
Fabriciooml/kafka-component@e58704971fd67645d3b448d9738b13293de741f8 -
Branch / Tag:
refs/tags/v0.1.0 - Owner: https://github.com/Fabriciooml
-
Access:
public
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
publish-pypi.yml@e58704971fd67645d3b448d9738b13293de741f8 -
Trigger Event:
release
-
Statement type:
File details
Details for the file kafka_component-0.1.0-py3-none-any.whl.
File metadata
- Download URL: kafka_component-0.1.0-py3-none-any.whl
- Upload date:
- Size: 6.9 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? Yes
- Uploaded via:
twine/7.0.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
dc3f54f87aea3f8637cffa17317f19faf9d1ad50b0cf9458617e2add97cd0526
|
|
| MD5 |
aad6fa82dbe88759eb9bc065b0654705
|
|
| BLAKE2b-256 |
df8be3f709aa546ecc9fd2759a5fe7e0801273e91138c2f6a71d1d4d2b3e00f2
|
Provenance
The following attestation bundles were made for kafka_component-0.1.0-py3-none-any.whl:
Publisher:
publish-pypi.yml on Fabriciooml/kafka-component
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
kafka_component-0.1.0-py3-none-any.whl -
Subject digest:
dc3f54f87aea3f8637cffa17317f19faf9d1ad50b0cf9458617e2add97cd0526 - Sigstore transparency entry: 2743034380
- Sigstore integration time:
-
Permalink:
Fabriciooml/kafka-component@e58704971fd67645d3b448d9738b13293de741f8 -
Branch / Tag:
refs/tags/v0.1.0 - Owner: https://github.com/Fabriciooml
-
Access:
public
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
publish-pypi.yml@e58704971fd67645d3b448d9738b13293de741f8 -
Trigger Event:
release
-
Statement type:
