feat: add nats support

.coveragerc

@@ -0,0 +1,4 @@
+[run]
+concurrency = multiprocessing
+parallel = true
+sigterm = true

example/README.md

@@ -102,3 +102,22 @@ Connect the app via Redis with:
 ```commandline
 asyncfast run amgi-redis main:app  input_channel
 ```
+
+## NATS
+
+Run the NATS compose file with:
+
+```commandline
+docker compose --file nats/docker-compose.yaml up --detach
+```
+
+This includes:
+
+- Redis `nats://localhost:4222/`
+- NUI NATS management GUI ([`http://localhost:31311/`](http://localhost:31311/))
+
+Connect the app via Redis with:
+
+```commandline
+asyncfast run amgi-nats-py-core main:app input_channel
+```

example/nats/docker-compose.yaml

@@ -0,0 +1,12 @@
+services:
+  nats:
+    image: nats:2.12.2
+    ports:
+      - "4222:4222"
+      - "8222:8222"
+  nui:
+    image: ghcr.io/nats-nui/nui:0.8.1
+    ports:
+      - 31311:31311
+    depends_on:
+      - nats

packages/amgi-aiobotocore/tests_amgi_aiobotocore/test_sqs_message_integration.py

@@ -4,10 +4,13 @@
 from uuid import uuid4
 
 import pytest
+from amgi_aiobotocore.sqs import _run_cli
+from amgi_aiobotocore.sqs import run
 from amgi_aiobotocore.sqs import Server
 from amgi_aiobotocore.sqs import SqsBatchFailureError
 from amgi_types import MessageAckEvent
 from amgi_types import MessageNackEvent
+from test_utils import assert_run_can_terminate
 from test_utils import MockApp
 from testcontainers.localstack import LocalStackContainer
 
@@ -288,3 +291,25 @@ async def test_lifespan(
                 "type": "message",
                 "state": {"item": state_item},
             }
+
+
+def test_run(queue_name: str, localstack_container: LocalStackContainer) -> None:
+    assert_run_can_terminate(
+        run,
+        queue_name,
+        region_name=localstack_container.region_name,
+        endpoint_url=localstack_container.get_url(),
+        aws_access_key_id="testcontainers-localstack",
+        aws_secret_access_key="testcontainers-localstack",
+    )
+
+
+def test_run_cli(queue_name: str, localstack_container: LocalStackContainer) -> None:
+    assert_run_can_terminate(
+        _run_cli,
+        [queue_name],
+        region_name=localstack_container.region_name,
+        endpoint_url=localstack_container.get_url(),
+        aws_access_key_id="testcontainers-localstack",
+        aws_secret_access_key="testcontainers-localstack",
+    )

packages/amgi-aiokafka/tests_amgi_aiokafka/test_kafka_message_integration.py

@@ -4,8 +4,11 @@
 import pytest
 from aiokafka import AIOKafkaConsumer
 from aiokafka import AIOKafkaProducer
+from amgi_aiokafka import _run_cli
+from amgi_aiokafka import run
 from amgi_aiokafka import Server
 from amgi_types import MessageAckEvent
+from test_utils import assert_run_can_terminate
 from test_utils import MockApp
 from testcontainers.kafka import KafkaContainer
 
@@ -156,3 +159,11 @@ async def test_lifespan(bootstrap_server: str, topic: str) -> None:
                 "type": "message",
                 "state": {"item": state_item},
             }
+
+
+def test_run(bootstrap_server: str, topic: str) -> None:
+    assert_run_can_terminate(run, topic, bootstrap_servers=bootstrap_server)
+
+
+def test_run_cli(bootstrap_server: str, topic: str) -> None:
+    assert_run_can_terminate(_run_cli, [topic], bootstrap_servers=bootstrap_server)

packages/amgi-common/src/amgi_common/__init__.py

@@ -266,15 +266,13 @@ def _remove_on_exception(self, key: H, task: Task[R]) -> None:
 
 
 class _Server(Protocol):
-    async def serve(self) -> None:
-        pass
+    async def serve(self) -> None: ...
 
-    def stop(self) -> None:
-        pass
+    def stop(self) -> None: ...
 
 
 def server_serve(server: _Server) -> None:
-    loop = asyncio.get_event_loop()
+    loop = asyncio.new_event_loop()
     loop.run_until_complete(_server_serve_async(server, loop))
 
 

packages/amgi-nats-py/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 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-nats-py/pyproject.toml

@@ -0,0 +1,40 @@
+[build-system]
+build-backend = "uv_build"
+requires = [ "uv-build>=0.8.14,<0.9.0" ]
+
+[project]
+name = "amgi-nats-py"
+version = "0.25.1"
+description = "Add your description here"
+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-common==0.25.1",
+  "amgi-types==0.25.1",
+  "nats-py>=2.12.0",
+]
+entry-points.amgi_server.amgi-nats-py-core = "amgi_nats_py.core:_run_cli"
+
+[dependency-groups]
+dev = [
+  "test-utils",
+  "testcontainers[nats]>=4.13.0",
+]
+
+[tool.uv.sources]
+amgi-common = { workspace = true }
+amgi-types = { workspace = true }
+test-utils = { workspace = true }

packages/amgi-nats-py/src/amgi_nats_py/_base.py

@@ -0,0 +1,87 @@
+import asyncio
+from abc import ABC
+from abc import abstractmethod
+from asyncio import Event
+from functools import partial
+from typing import Any
+from typing import Union
+
+from amgi_common import Lifespan
+from amgi_types import AMGIApplication
+from amgi_types import AMGISendCallable
+from amgi_types import MessageReceiveEvent
+from amgi_types import MessageScope
+from amgi_types import MessageSendEvent
+from nats.aio.client import Client
+from nats.aio.msg import Msg
+from nats.js import JetStreamContext
+
+
+NatsClient = Union[Client, JetStreamContext]
+
+
+class _Receive:
+    def __init__(self, msg: Msg) -> None:
+        self._msg = msg
+
+    async def __call__(self) -> MessageReceiveEvent:
+        return {
+            "type": "message.receive",
+            "id": self._msg.reply or "",
+            "headers": [
+                (key.encode(), value.encode())
+                for key, value in (self._msg.headers or {}).items()
+            ],
+            "payload": self._msg.data,
+        }
+
+
+async def _message_send(client: NatsClient, event: MessageSendEvent) -> None:
+    await client.publish(
+        event["address"],
+        event.get("payload") or b"",
+        headers={key.decode(): value.decode() for key, value in event["headers"]},
+    )
+
+
+class _BaseServer(ABC):
+    def __init__(
+        self, app: AMGIApplication, *subjects: str, servers: str | list[str]
+    ) -> None:
+        self._app = app
+        self._subjects = subjects
+        self._servers = servers
+
+        self._stop_event = Event()
+        self._state: dict[str, Any] = {}
+
+    async def _client_serve(self, client: NatsClient) -> None:
+        client_subscribe_cb = partial(self._subscribe_cb, client)
+        subscriptions = await asyncio.gather(
+            *(
+                client.subscribe(subject, cb=client_subscribe_cb)
+                for subject in self._subjects
+            )
+        )
+
+        async with Lifespan(self._app, self._state):
+            await self._stop_event.wait()
+
+            for subscription in subscriptions:
+                await subscription.unsubscribe()
+
+    async def _subscribe_cb(self, client: NatsClient, msg: Msg) -> None:
+        scope: MessageScope = {
+            "type": "message",
+            "amgi": {"version": "1.0", "spec_version": "1.0"},
+            "address": msg.subject,
+            "state": self._state.copy(),
+        }
+        await self._app(scope, _Receive(msg), self._send(client, msg))
+
+    def stop(self) -> None:
+        self._stop_event.set()
+
+    @abstractmethod
+    def _send(self, client: NatsClient, msg: Msg) -> AMGISendCallable:
+        raise NotImplementedError  # pragma: no cover

packages/amgi-nats-py/src/amgi_nats_py/core.py

@@ -0,0 +1,50 @@
+from amgi_common import server_serve
+from amgi_nats_py._base import _BaseServer
+from amgi_nats_py._base import _message_send
+from amgi_nats_py._base import NatsClient
+from amgi_types import AMGIApplication
+from amgi_types import AMGISendCallable
+from amgi_types import AMGISendEvent
+from nats.aio.client import Client
+from nats.aio.msg import Msg
+
+
+def run(app: AMGIApplication, *subjects: str, servers: str | list[str]) -> None:
+    server = Server(app, *subjects, servers=servers)
+    server_serve(server)
+
+
+def _run_cli(
+    app: AMGIApplication,
+    subjects: list[str],
+    servers: list[str] | None = None,
+) -> None:
+    if servers is None:
+        servers = ["nats://localhost:4222"]
+    run(app, *subjects, servers=servers)
+
+
+class _Send:
+    def __init__(self, client: Client, msg: Msg) -> None:
+        self._client = client
+        self._msg = msg
+
+    async def __call__(self, event: AMGISendEvent) -> None:
+        if event["type"] == "message.send":
+            await _message_send(self._client, event)
+
+
+class Server(_BaseServer):
+    def __init__(
+        self, app: AMGIApplication, *subjects: str, servers: str | list[str]
+    ) -> None:
+        super().__init__(app, *subjects, servers=servers)
+
+    async def serve(self) -> None:
+        client = Client()
+        await client.connect(self._servers)
+
+        await self._client_serve(client)
+
+    def _send(self, client: NatsClient, msg: Msg) -> AMGISendCallable:
+        return _Send(client, msg)

packages/amgi-nats-py/tests_amgi_nats/conftest.py

@@ -0,0 +1,18 @@
+from collections.abc import Generator
+
+import pytest
+from nats.aio.client import Client
+from testcontainers.nats import NatsContainer
+
+
+@pytest.fixture(scope="package")
+def nats_container() -> Generator[NatsContainer, None, None]:
+    with NatsContainer() as nats_container:
+        yield nats_container
+
+
+@pytest.fixture
+async def client(nats_container: NatsContainer) -> Client:
+    client = Client()
+    await client.connect(nats_container.nats_uri())
+    return client

packages/amgi-nats-py/tests_amgi_nats/test_nats_core_message.py

@@ -0,0 +1,14 @@
+from unittest.mock import Mock
+from unittest.mock import patch
+
+from amgi_nats_py import core
+from amgi_nats_py.core import _run_cli
+
+
+def test_run_cli_defaults() -> None:
+    with patch.object(core, "run") as mock_run:
+        mock_app = Mock()
+        _run_cli(mock_app, ["subject"])
+        mock_run.assert_called_once_with(
+            mock_app, "subject", servers=["nats://localhost:4222"]
+        )