✨ SQS server

Author: aleksul

pyproject.toml

@@ -14,7 +14,8 @@ docs = [
 lint = [
   "mypy>=1.19.1",
   "ruff>=0.15.8",
-  "types-grpcio>=1.0.0.20251009"
+  "types-grpcio>=1.0.0.20251009",
+  "types-aiobotocore[sqs]>=3.4.0"
 ]
 pre-commit = [
   "pre-commit>=4.5.1"
@@ -64,6 +65,7 @@ pubsub = [
 ]
 pydantic = ["pydantic>=2.0.0,<3.0.0"]
 redis = ["redis<8.0.0,>=7.0.0"]
+sqs = ["aiobotocore>=2.10.0,<3.0.0"]
 
 [project.urls]
 documentation = "https://repid.aleksul.space"

repid/connections/__init__.py

@@ -53,3 +53,8 @@
     from repid.connections.nats import NatsServer
 
     __all__ += ["NatsServer"]
+
+if is_installed("aiobotocore"):
+    from repid.connections.sqs import SqsServer
+
+    __all__ += ["SqsServer"]

repid/connections/sqs/__init__.py

@@ -0,0 +1,3 @@
+from repid.connections.sqs.message_broker import SqsServer
+
+__all__ = ["SqsServer"]

repid/connections/sqs/message.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import base64
+import logging
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Any
+
+from repid.connections.abc import MessageAction, ReceivedMessageT
+
+if TYPE_CHECKING:
+    from repid.connections.sqs.message_broker import SqsServer
+
+logger = logging.getLogger("repid.connections.sqs")
+
+
+class SqsReceivedMessage(ReceivedMessageT):
+    def __init__(
+        self,
+        server: SqsServer,
+        channel: str,
+        queue_url: str,
+        msg: Mapping[str, Any],
+    ) -> None:
+        self._server = server
+        self._channel = channel
+        self._queue_url = queue_url
+        self._msg = msg
+        self._action: MessageAction | None = None
+
+        self._message_id = msg.get("MessageId")
+        self._receipt_handle = msg.get("ReceiptHandle")
+
+        self._headers: dict[str, str] = {}
+        self._content_type: str | None = None
+
+        attributes = msg.get("MessageAttributes", {})
+        for key, value in attributes.items():
+            if key == "content-type":
+                self._content_type = value.get("StringValue")
+            else:
+                self._headers[key] = value.get("StringValue")
+
+        body = str(msg.get("Body", ""))
+        self._payload = base64.b64decode(body)
+
+    @property
+    def payload(self) -> bytes:
+        return self._payload
+
+    @property
+    def headers(self) -> dict[str, str] | None:
+        return self._headers
+
+    @property
+    def content_type(self) -> str | None:
+        return self._content_type
+
+    @property
+    def channel(self) -> str:
+        return self._channel
+
+    @property
+    def action(self) -> MessageAction | None:
+        return self._action
+
+    @property
+    def is_acted_on(self) -> bool:
+        return self._action is not None
+
+    @property
+    def message_id(self) -> str | None:
+        return self._message_id
+
+    async def ack(self) -> None:
+        if self._action is not None:
+            return
+        self._action = MessageAction.acked
+
+        if self._server._client is not None and self._receipt_handle:
+            await self._server._client.delete_message(
+                QueueUrl=self._queue_url,
+                ReceiptHandle=self._receipt_handle,
+            )
+
+    async def nack(self) -> None:
+        if self._action is not None:
+            return
+        self._action = MessageAction.nacked
+
+        if self._server._client is not None and self._receipt_handle:
+            dlq_strategy = self._server._dlq_queue_strategy
+            if dlq_strategy:
+                dlq_channel = dlq_strategy(self._channel)
+                dlq_queue_url = await self._server._get_queue_url(dlq_channel)
+
+                message_attributes: dict[str, Any] = {}
+                for k, v in self._headers.items():
+                    message_attributes[k] = {"DataType": "String", "StringValue": v}
+                if self._content_type:
+                    message_attributes["content-type"] = {
+                        "DataType": "String",
+                        "StringValue": self._content_type,
+                    }
+
+                await self._server._client.send_message(
+                    QueueUrl=dlq_queue_url,
+                    MessageBody=self._msg.get("Body", ""),
+                    MessageAttributes=message_attributes,
+                )
+
+            # SQS needs delete_message after NACK if moved to DLQ,
+            # or VisibilityTimeout=0 if we don't handle DLQ natively and just want redelivery.
+            # But the plan specifies "followed by an ack() on the original queue".
+            await self._server._client.delete_message(
+                QueueUrl=self._queue_url,
+                ReceiptHandle=self._receipt_handle,
+            )
+
+    async def reject(self) -> None:
+        if self._action is not None:
+            return
+        self._action = MessageAction.rejected
+
+        if self._server._client is not None and self._receipt_handle:
+            await self._server._client.change_message_visibility(
+                QueueUrl=self._queue_url,
+                ReceiptHandle=self._receipt_handle,
+                VisibilityTimeout=0,
+            )
+
+    async def reply(
+        self,
+        *,
+        payload: bytes,
+        headers: dict[str, str] | None = None,
+        content_type: str | None = None,
+        channel: str | None = None,
+        server_specific_parameters: dict[str, Any] | None = None,
+    ) -> None:
+        if self._action is not None:
+            return
+        self._action = MessageAction.replied
+
+        if self._server._client is not None and self._receipt_handle:
+            reply_channel = channel if channel is not None else self._channel
+            reply_queue_url = await self._server._get_queue_url(reply_channel)
+
+            message_attributes: dict[str, Any] = {}
+            if headers:
+                for k, v in headers.items():
+                    message_attributes[k] = {"DataType": "String", "StringValue": v}
+            if content_type:
+                message_attributes["content-type"] = {
+                    "DataType": "String",
+                    "StringValue": content_type,
+                }
+
+            kwargs: dict[str, Any] = {
+                "QueueUrl": reply_queue_url,
+                "MessageBody": base64.b64encode(payload).decode("ascii") or " ",
+                "MessageAttributes": message_attributes,
+            }
+            if server_specific_parameters:
+                kwargs.update(server_specific_parameters)
+
+            await self._server._client.send_message(**kwargs)
+
+            await self._server._client.delete_message(
+                QueueUrl=self._queue_url,
+                ReceiptHandle=self._receipt_handle,
+            )