draft: SQS integration.

Author: Carlos Bellino

faststream/sqs/__init__.py

@@ -0,0 +1,24 @@
+from faststream.broker.test import TestApp
+from faststream.sqs.annotations import SQSBroker, SQSMessage, SQSProducer
+from faststream.sqs.router import SQSRouter
+from faststream.sqs.shared.router import SQSRoute
+from faststream.sqs.shared.schemas import (
+    FifoQueue,
+    RedriveAllowPolicy,
+    RedrivePolicy,
+    SQSQueue,
+)
+from faststream.sqs.test import TestSQSBroker
+
+__all__ = (
+    "FifoQueue",
+    "RedriveAllowPolicy",
+    "RedrivePolicy",
+    "SQSBroker",
+    "SQSMessage",
+    "SQSQueue",
+    "SQSRouter",
+    "SQSRoute",
+    "TestApp",
+    "TestSQSBroker",
+)

faststream/sqs/annotations.py

@@ -0,0 +1,19 @@
+from faststream._compat import Annotated
+from faststream.annotations import ContextRepo, Logger, NoCast
+from faststream.sqs.broker import SQSBroker as SB  # NOQA
+from faststream.sqs.message import SQSMessage as SM  # NOQA
+from faststream.sqs.producer import SQSFastProducer
+from faststream.utils.context import Context
+
+__all__ = (
+    "Logger",
+    "ContextRepo",
+    "NoCast",
+    "SQSBroker",
+    "SQSMessage",
+    "SQSProducer",
+)
+
+SQSBroker = Annotated[SB, Context("broker")]
+SQSMessage = Annotated[SM, Context("message")]
+SQSProducer = Annotated[SQSFastProducer, Context("broker._producer")]

faststream/sqs/asyncapi.py

@@ -0,0 +1,8 @@
+class Publisher:
+    # TODO
+    pass
+
+
+class Handler:
+    # TODO
+    pass

faststream/sqs/broker.py

@@ -0,0 +1,3 @@
+class SQSBroker:
+    # TODO
+    pass

faststream/sqs/handler.py

@@ -0,0 +1,10 @@
+from aiobotocore.client import AioBaseClient
+
+from faststream.broker.handler import AsyncHandler
+from faststream.types import AnyDict
+
+
+class LogicSQSHandler(AsyncHandler[AnyDict]):
+    async def start(self, client: AioBaseClient) -> None:
+        # TODO check "start" method on broker
+        pass

faststream/sqs/message.py

@@ -0,0 +1,26 @@
+from typing import Any
+
+from faststream.broker.message import StreamMessage
+from faststream.types import AnyDict
+
+
+class SQSMessage(StreamMessage[AnyDict]):
+    def __init__(
+        self,
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(*args, **kwargs)
+        self.commited = False
+
+    async def ack(self, **kwargs: Any) -> None:
+        # TODO: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html
+        self.commited = True
+
+    async def nack(self, **kwargs: Any) -> None:
+        # TODO
+        self.commited = True
+
+    async def reject(self, **kwargs: Any) -> None:
+        # TODO
+        self.commited = True

faststream/sqs/producer.py

@@ -0,0 +1,3 @@
+class SQSFastProducer:
+    # TODO
+    pass

faststream/sqs/router.py

@@ -0,0 +1,6 @@
+from faststream.sqs.shared.router import SQSRouter as BaseRouter
+
+
+class SQSRouter(BaseRouter):
+    # TODO
+    pass

faststream/sqs/shared/__init__.py

@@ -0,0 +1,56 @@
+import logging
+from typing import Any, Optional
+
+from faststream._compat import override
+from faststream.broker.core.mixins import LoggingMixin
+from faststream.broker.message import StreamMessage
+from faststream.log import access_logger
+from faststream.types import AnyDict
+
+
+class SQSLoggingMixin(LoggingMixin):
+    _max_queue_len: int
+
+    def __init__(
+        self,
+        *args: Any,
+        logger: Optional[logging.Logger] = access_logger,
+        log_level: int = logging.INFO,
+        log_fmt: Optional[str] = None,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(
+            *args,
+            logger=logger,
+            log_level=log_level,
+            log_fmt=log_fmt,
+            **kwargs,
+        )
+        self._max_queue_len = 4
+
+    @override
+    def _get_log_context(  # type: ignore[override]
+        self,
+        message: Optional[StreamMessage[Any]],
+        queue: str = "",
+    ) -> AnyDict:
+        return {
+            "queue": queue,
+            **super()._get_log_context(message),
+        }
+
+    @property
+    def fmt(self) -> str:
+        return self._fmt or (
+            "%(asctime)s %(levelname)s - "
+            f"%(queue)-{self._max_queue_len}s | "
+            "%(message_id)-10s "
+            "- %(message)s"
+        )
+
+    def _setup_log_context(
+        self,
+        queue: Optional[str] = None,
+    ) -> None:
+        if queue is not None:
+            self._max_queue_len = max((self._max_queue_len, len(queue)))