feat(application): use buffers instead of logs

Author: emptybutton

README.md

@@ -5,7 +5,7 @@
 [![Lines](https://img.shields.io/endpoint?url=https%3A%2F%2Fghloc.vercel.app%2Fapi%2Femptybutton%2Ftgdb%2Fbadge%3Ffilter%3D.py&logo=python&label=lines&color=blue)](https://github.com/search?q=repo%3Aemptybutton%2tgdb+language%3APython+&type=code)
 [![Wakatime](https://wakatime.com/badge/user/0d3b7ff5-0547-4323-a43e-2a7308d973a0/project/2e316b92-fcf1-44d8-ad77-6c81e23cdfe2.svg)](https://wakatime.com/badge/user/0d3b7ff5-0547-4323-a43e-2a7308d973a0/project/2e316b92-fcf1-44d8-ad77-6c81e23cdfe2)
 
-ACID NoSQL РСУБД для хранения данных в телеграме.
+ACID NoSQL РСУБД поверх телеграма.
 
 ```bash
 pip install tgdb

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "tgdb"
 version = "0.1.0"
-description = "ACID NoSQL РСУБД для хранения данных в телеграме"
+description = "ACID NoSQL РСУБД поверх телеграма"
 authors = [
     {name = "Alexander Smolin", email = "[email protected]"}
 ]
@@ -15,6 +15,7 @@ dependencies = [
     "typenv>=0.2.0",
     "dishka>=1.4.2",
     "fastapi>=0.115.2",
+    "pydantic>=2.11.4",
     "uvicorn[standard]>=0.30.6",
 ]
 

src/tgdb/application/output_commits.py

@@ -1,19 +1,17 @@
-from collections.abc import AsyncIterable
+from collections.abc import Sequence
 from dataclasses import dataclass
 
-from tgdb.application.ports.commit_serialization import CommitSerialization
+from tgdb.application.ports.buffer import Buffer
 from tgdb.application.ports.queque import Queque
 from tgdb.entities.transaction import TransactionCommit
 
 
-class InvalidOperatorError(Exception): ...
-
-
 @dataclass(frozen=True)
-class OutputCommits[SerializedCommitT]:
-    output_commits: Queque[TransactionCommit]
-    commit_serialization: CommitSerialization[SerializedCommitT]
+class OutputCommits:
+    commit_buffer: Buffer[TransactionCommit]
+    output_commits: Queque[Sequence[TransactionCommit]]
 
-    async def __call__(self) -> AsyncIterable[SerializedCommitT]:
-        async for commit in self.output_commits:
-            yield await self.commit_serialization.serialized(commit)
+    async def __call__(self) -> None:
+        async for commits in self.commit_buffer:
+            await self.output_commits.push(commits)
+            await self.output_commits.sync()

src/tgdb/application/output_commits_to_clients.py

@@ -0,0 +1,19 @@
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from tgdb.application.ports.notification import Notification
+from tgdb.application.ports.queque import Queque
+from tgdb.entities.transaction import TransactionCommit
+
+
+class InvalidOperatorError(Exception): ...
+
+
+@dataclass(frozen=True)
+class OutputCommitsToClients:
+    output_commits: Queque[Sequence[TransactionCommit]]
+    notification: Notification[Sequence[TransactionCommit]]
+
+    async def __call__(self) -> None:
+        async for commits in self.output_commits:
+            await self.notification.send(commits)

src/tgdb/application/output_commits_to_heap.py

@@ -1,22 +1,28 @@
+from collections.abc import Sequence
 from dataclasses import dataclass
 
 from tgdb.application.ports.heap import Heap
-from tgdb.application.ports.message import TransactionCommitMessage
 from tgdb.application.ports.queque import Queque
-from tgdb.entities.transaction import TransactionOkCommit
+from tgdb.entities.transaction import TransactionCommit, TransactionOkCommit
 
 
 @dataclass(frozen=True)
 class OutputCommitsToHeap:
     heap: Heap
-    output_commit_messages: Queque[TransactionCommitMessage]
+    output_commits: Queque[Sequence[TransactionCommit]]
 
     async def __call__(self) -> None:
-        async for message in self.output_commit_messages:
-            if not isinstance(message.commit, TransactionOkCommit):
-                return
+        is_previous_map_partial = True
 
-            if message.is_commit_duplicate:
-                await self.heap.map_as_duplicate(message.commit.effect)
+        async for commits in self.output_commits:
+            effects = tuple(
+                commit.effect
+                for commit in commits
+                if isinstance(commit, TransactionOkCommit)
+            )
+
+            if is_previous_map_partial:
+                await self.heap.map_as_duplicate(effects)
+                is_previous_map_partial = False
             else:
-                await self.heap.map(message.commit.effect)
+                await self.heap.map(effects)

src/tgdb/application/output_commits_to_log.py

@@ -0,0 +1,7 @@
+from abc import ABC, abstractmethod
+from collections.abc import AsyncIterable, Sequence
+
+
+class Buffer[ValueT](ABC, AsyncIterable[Sequence[ValueT]]):
+    @abstractmethod
+    async def add(self, value: ValueT, /) -> None: ...

src/tgdb/application/ports/buffer.py

@@ -1,11 +1,14 @@
 from abc import ABC, abstractmethod
+from collections.abc import Sequence
 
 from tgdb.entities.transaction import TransactionEffect
 
 
 class Heap(ABC):
     @abstractmethod
-    async def map(self, effect: TransactionEffect, /) -> None: ...
+    async def map(self, effects: Sequence[TransactionEffect], /) -> None: ...
 
     @abstractmethod
-    async def map_as_duplicate(self, effect: TransactionEffect, /) -> None: ...
+    async def map_as_duplicate(
+        self, effects: Sequence[TransactionEffect], /
+    ) -> None: ...