feat: add sqla broker

Author: Arseniy-Popov

faststream/sqla/message.py

@@ -46,21 +46,21 @@ def __init__(
 
         super().__init__(raw_message=self, body=payload)
 
-    def mark_completed(self) -> None:
+    def _mark_completed(self) -> None:
         self.state = SqlaMessageState.COMPLETED
         self.next_attempt_at = None
         self.to_archive = True
 
-    def mark_retryable(self, *, next_attempt_at: datetime) -> None:
+    def _mark_retryable(self, *, next_attempt_at: datetime) -> None:
         self.state = SqlaMessageState.RETRYABLE
         self.next_attempt_at = next_attempt_at
 
-    def mark_failed(self) -> None:
+    def _mark_failed(self) -> None:
         self.state = SqlaMessageState.FAILED
         self.next_attempt_at = None
         self.to_archive = True
 
-    def mark_pending(self) -> None:
+    def _mark_pending(self) -> None:
         self.state = SqlaMessageState.PENDING
         self.acquired_at = None
 
@@ -72,15 +72,15 @@ def allow_attempt(self) -> bool:
             first_attempt_at=self.first_attempt_at,
             attempts_count=self.attempts_count,
         ):
-            self.mark_failed()
+            self._mark_failed()
             return False
         return True
 
     async def ack(self) -> None:
         if self.decision_recorded:
             return
 
-        self.mark_completed()
+        self._mark_completed()
         self.decision_recorded = True
 
     async def nack(self) -> None:
@@ -94,14 +94,14 @@ async def nack(self) -> None:
                 attempts_count=self.attempts_count,
             )
         ):
-            self.mark_failed()
+            self._mark_failed()
         else:
-            self.mark_retryable(next_attempt_at=next_attempt_at)
+            self._mark_retryable(next_attempt_at=next_attempt_at)
         self.decision_recorded = True
 
     async def reject(self) -> None:
         if self.decision_recorded:
             return
 
-        self.mark_failed()
+        self._mark_failed()
         self.decision_recorded = True

faststream/sqla/subscriber/usecase.py

@@ -67,7 +67,7 @@ async def stop(self) -> None:
         drained = self._drain_acquired()
         if drained:
             for message in drained:
-                message.mark_pending()
+                message._mark_pending()
             self._buffer_results(drained)
             self.add_task(self._flush_results)
         with suppress(asyncio.TimeoutError):