feat: add sqla broker

Author: Arseniy-Popov

faststream/sqla/__init__.py

@@ -0,0 +1,67 @@
+import logging
+from typing import Any, Iterable, Optional, Union
+
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+from faststream._internal.basic_types import LoggerProto
+from faststream._internal.broker import BrokerUsecase
+from faststream._internal.constants import EMPTY
+from faststream.security import BaseSecurity
+from faststream.specification.schema.broker import BrokerSpec
+from faststream.specification.schema.extra.tag import Tag, TagDict
+from faststream.sqla.broker.registrator import SqlaRegistrator
+from faststream.sqla.configs.broker import SqlaBrokerConfig
+from faststream.sqla.broker.logging import make_sqla_logger_state
+
+
+class SqlaBroker(
+    SqlaRegistrator,
+    BrokerUsecase[
+        Any,
+        Any,
+    ],
+):
+    url: list[str]
+
+    def __init__(
+        self,
+        *,
+        engine: AsyncEngine,
+        # broker base args
+        routers: Iterable[SqlaRegistrator] = (),
+        # AsyncAPI args
+        security: Optional["BaseSecurity"] = None,
+        specification_url: str | Iterable[str] | None = None,
+        protocol: str | None = None,
+        protocol_version: str | None = "auto",
+        description: str | None = None,
+        tags: Iterable[Union["Tag", "TagDict"]] = (),
+        # logging args
+        logger: Optional["LoggerProto"] = EMPTY,
+        log_level: int = logging.INFO,
+    ) -> None:
+
+        super().__init__(
+            routers=routers,
+            config=SqlaBrokerConfig(
+                logger=make_sqla_logger_state(
+                    logger=logger,
+                    log_level=log_level,
+                ),
+            ),
+            specification=BrokerSpec(
+                description=description,
+                url=specification_url,
+                protocol=protocol,
+                protocol_version=protocol_version,
+                security=security,
+                tags=tags,
+            ),
+        )
+    
+    async def _connect(self) -> Any:
+        return True
+
+    async def start(self) -> None:
+        await self.connect()
+        await super().start()

faststream/sqla/broker/__init__.py

@@ -0,0 +1,50 @@
+import logging
+from functools import partial
+from typing import TYPE_CHECKING, Any
+
+from faststream._internal.logger import DefaultLoggerStorage, make_logger_state
+from faststream._internal.logger.logging import get_broker_logger
+
+if TYPE_CHECKING:
+    from faststream._internal.basic_types import LoggerProto
+    from faststream._internal.context import ContextRepo
+
+
+class SqlaParamsStorage(DefaultLoggerStorage):
+    def __init__(self) -> None:
+        super().__init__()
+
+        self.logger_log_level = logging.INFO
+
+    def set_level(self, level: int) -> None:
+        self.logger_log_level = level
+
+    def register_subscriber(self, params: dict[str, Any]) -> None:
+        return
+
+    def get_logger(self, *, context: "ContextRepo") -> "LoggerProto":
+        message_id_ln = 10
+
+        # TODO: generate unique logger names to not share between brokers
+        if not (lg := self._get_logger_ref()):
+            lg = get_broker_logger(
+                name="sqla",
+                default_context={},
+                message_id_ln=message_id_ln,
+                fmt="".join((
+                    "%(asctime)s %(levelname)-8s - ",
+                    f"%(message_id)-{message_id_ln}s ",
+                    "- %(message)s",
+                )),
+                context=context,
+                log_level=self.logger_log_level,
+            )
+            self._logger_ref.add(lg)
+
+        return lg
+
+
+make_sqla_logger_state = partial(
+    make_logger_state,
+    default_storage_cls=SqlaParamsStorage,
+)

faststream/sqla/broker/broker.py

@@ -0,0 +1,55 @@
+from typing import Any, cast
+
+from sqlalchemy.ext.asyncio import AsyncEngine
+from faststream._internal.broker.registrator import Registrator
+from faststream._internal.constants import EMPTY
+from faststream.middlewares.acknowledgement.config import AckPolicy
+from faststream.sqla.configs.broker import SqlaBrokerConfig
+from faststream.sqla.subscriber.factory import create_subscriber
+from faststream.sqla.retry import RetryStrategy
+
+
+class SqlaRegistrator(Registrator[Any, Any]):
+    def subscriber(
+        self,
+        engine: AsyncEngine,
+        queue: str,
+        max_workers: int,
+        retry_strategy: RetryStrategy,
+        fetch_interval: float,
+        fetch_batch_size: int,
+        overfetch_factor: float,
+        flush_interval: float,
+        release_stuck_interval: float,
+        graceful_shutdown_timeout: float,
+        release_stuck_timeout: int,
+        ack_policy: AckPolicy = AckPolicy.NACK_ON_ERROR,
+    ) -> Any:
+        workers = max_workers or 1
+
+        subscriber = create_subscriber(
+            engine=engine,
+            queue=queue,
+            max_workers=max_workers,
+            retry_strategy=retry_strategy,
+            fetch_interval=fetch_interval,
+            fetch_batch_size=fetch_batch_size,
+            overfetch_factor=overfetch_factor,
+            flush_interval=flush_interval,
+            release_stuck_interval=release_stuck_interval,
+            graceful_shutdown_timeout=graceful_shutdown_timeout,
+            release_stuck_timeout=release_stuck_timeout,
+            config=cast("SqlaBrokerConfig", self.config),
+            ack_policy=ack_policy,
+        )
+
+        super().subscriber(subscriber)
+
+        # subscriber.add_call(
+        #     parser_=parser,
+        #     decoder_=decoder,
+        #     dependencies_=dependencies,
+        #     middlewares_=middlewares,
+        # )
+
+        return subscriber