amgi-kafka-event-source-mapping
amgi-kafka-event-source-mapping is an adaptor for AMGI applications to run in a Kafka event source mapped environment.
Installation
pip install amgi-kafka-event-source-mapping==0.50.0
Example
This example uses AsyncFast:
from dataclasses import dataclass
from amgi_kafka_event_source_mapping import KafkaEventSourceMappingHandler
from asyncfast import AsyncFast
app = AsyncFast()
@dataclass
class Order:
item_ids: list[str]
@app.channel("orders")
async def orders(order: Order) -> None:
# Makes an order
...
handler = KafkaEventSourceMappingHandler(app)
What it does
- Converts Kafka batch events into AMGI
message.receiveevents - Uses the Kafka topic name as the AMGI message address
- Supports partial batch failures so only failed records are reported
- Sends outbound messages to Kafka using an async producer
- Outbound messages are sent via the same Kafka broker (bootstrap servers) that the records were received from
- Optionally manages application startup and shutdown via AMGI lifespan
Record handling
- Record values and keys are passed to your app as bytes
- Kafka record headers become AMGI headers
- Records are only acknowledged when your app emits
message.ack - Records that emit
message.nackor are not acknowledged are treated as failures
Nack handling
By default, records that are negatively acknowledged, or not acknowledged are logged:
handler = KafkaEventSourceMappingHandler(app, on_nack="log")
To fail the invocation when any record is nacked, configure the handler to raise an error instead:
handler = KafkaEventSourceMappingHandler(app, on_nack="error")
This is useful when running in environments where a failed invocation should trigger a retry, or alert.
When using this mode, handlers must be idempotent. Kafka event source mappings may re-deliver records after failures, restarts, or rebalances, and your application logic should be safe to execute more than once for the same record.
Lifespan
Lifespan support is enabled by default.
- Startup runs once per Lambda execution environment
- Shutdown is attempted when the environment is terminated
Shutdown handling relies on signal.SIGTERM, which is supported by Python 3.12 and later Lambda runtimes.
To use fully stateless, per-invocation behavior, disable lifespan:
handler = KafkaEventSourceMappingHandler(app, lifespan=False)
Contact
For questions or suggestions, please contact jack.burridge@mail.com.
License
Copyright 2026 AMGI
1 maintainer
Metadata
1 maintainer
Release files for amgi-kafka-event-source-mapping 0.50.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| amgi_kafka_event_source_mapping-0.50.0-py3-none-any.whl | Python 3 | none | any | Details |
Release files / amgi_kafka_event_source_mapping-0.50.0-py3-none-any.whl
| Download URL | amgi_kafka_event_source_mapping-0.50.0-py3-none-any.whl |
|---|---|
| Size | 6.8 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
67aa0f8e005b75a49cc9114553c5beb8b8250bda9b73ce9aab7bdba96b81adcf
|
|
BLAKE2b-256 checksum How to use checksums |
0e04b73c319abef7805cd885fe10772cf680960c8ce0720608831dd1c34ebc1b
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.4 {"installer":{"name":"uv","version":"0.13.4","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}
|
Security for amgi-kafka-event-source-mapping 0.50.0
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
