Feat/add kafka event source mapping support

packages/amgi-aiokafka/src/amgi_aiokafka/__init__.py

@@ -1,7 +1,6 @@
 import asyncio
 import logging
 import sys
-from asyncio import Lock
 from collections import deque
 from collections.abc import Awaitable
 from collections.abc import Callable
@@ -119,18 +118,17 @@ async def __call__(self, event: AMGISendEvent) -> None:
 class MessageSend:
     def __init__(self, bootstrap_servers: str | list[str]) -> None:
         self._bootstrap_servers = bootstrap_servers
-        self._producer = None
-        self._producer_lock = Lock()
 
     async def __aenter__(self) -> Self:
+        self._producer = AIOKafkaProducer(bootstrap_servers=self._bootstrap_servers)
+        await self._producer.start()
         return self
 
     async def __call__(self, event: MessageSendEvent) -> None:
-        producer = await self._get_producer()
         encoded_headers = [(key.decode(), value) for key, value in event["headers"]]
 
         key = event.get("bindings", {}).get("kafka", {}).get("key")
-        await producer.send(
+        await self._producer.send_and_wait(
             event["address"],
             headers=encoded_headers,
             value=event.get("payload"),
@@ -143,19 +141,7 @@ async def __aexit__(
         exc_val: BaseException | None,
         exc_tb: TracebackType | None,
     ) -> None:
-        if self._producer is not None:
-            await self._producer.stop()
-
-    async def _get_producer(self) -> AIOKafkaProducer:
-        if self._producer is None:
-            async with self._producer_lock:
-                if self._producer is None:
-                    producer = AIOKafkaProducer(
-                        bootstrap_servers=self._bootstrap_servers
-                    )
-                    await producer.start()
-                    self._producer = producer
-        return self._producer
+        await self._producer.stop()
 
 
 class Server:

packages/amgi-kafka-event-source-mapping/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Jack Burridge
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

packages/amgi-kafka-event-source-mapping/README.md

@@ -0,0 +1,96 @@
+# amgi-kafka-event-source-mapping
+
+amgi-kafka-event-source-mapping is an adaptor for [AMGI](https://amgi.readthedocs.io/en/latest/) applications to run in
+a Kafka event source mapped environment.
+
+## Installation
+
+```bash
+pip install amgi-kafka-event-source-mapping==0.32.0
+```
+
+## Example
+
+This example uses [AsyncFast](https://pypi.org/project/asyncfast/):
+
+```python
+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.receive` events
+- 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.nack` or are not acknowledged are treated as failures
+
+## Nack handling
+
+By default, records that are negatively acknowledged, or not acknowledged are logged:
+
+```python
+handler = KafkaEventSourceMappingHandler(app, on_nack="log")
+```
+
+To fail the invocation when any record is nacked, configure the handler to raise an error instead:
+
+```python
+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:
+
+```python
+handler = KafkaEventSourceMappingHandler(app, lifespan=False)
+```
+
+## Contact
+
+For questions or suggestions, please contact [jack.burridge@mail.com](mailto:jack.burridge@mail.com).
+
+## License
+
+Copyright 2026 AMGI

packages/amgi-kafka-event-source-mapping/pyproject.toml

@@ -0,0 +1,49 @@
+[build-system]
+build-backend = "uv_build"
+requires = [
+  "uv-build>=0.9.26,<0.10.0",
+]
+
+[project]
+name = "amgi-kafka-event-source-mapping"
+version = "0.32.0"
+description = "Kafka event source mapping adaptor for AMGI applications"
+readme = "README.md"
+license = "MIT"
+license-files = [
+  "LICENSE",
+]
+authors = [
+  { name = "jack.burridge", email = "jack.burridge@mail.com" },
+]
+requires-python = ">=3.10"
+classifiers = [
+  "Programming Language :: Python :: 3 :: Only",
+  "Programming Language :: Python :: 3.10",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  "Programming Language :: Python :: 3.13",
+  "Programming Language :: Python :: 3.14",
+]
+dependencies = [
+  "amgi-aiokafka==0.32.0",
+  "typing-extensions>=4.15.0; python_full_version<'3.11'",
+
+]
+
+[dependency-groups]
+dev = [
+  "pytest>=8.4.1",
+  "pytest-asyncio>=1.3.0",
+  "pytest-cov>=7.0.0",
+  "pytest-timeout>=2.4.0",
+  "test-utils",
+  "testcontainers[kafka]>=4.13.0",
+
+]
+
+[tool.uv.sources.amgi-aiokafka]
+workspace = true
+
+[tool.uv.sources.test-utils]
+workspace = true