feat(infrastructure): add a part for adapters

Author: emptybutton

src/tgdb/application/common/ports/channel.py

@@ -26,4 +26,4 @@ async def publish(
     ) -> None: ...
 
     @abstractmethod
-    async def wait(self, xid: XID) -> Notification: ...
+    async def wait(self, xid: XID, /) -> Notification: ...

src/tgdb/entities/relation/relation.py

@@ -32,13 +32,9 @@ class RelationVersionID:
 class NotIncrementedRelationVersionError(Exception): ...
 
 
-class RelationWithoutLastVersionError(Exception): ...
-
-
 @dataclass
 class Relation:
     """
-    :raises tgdb.entities.relation.relation.RelationWithoutLastVersionError:
     :raises tgdb.entities.relation.relation.NotIncrementedRelationVersionError:
     """
 
@@ -47,9 +43,6 @@ class Relation:
     _intermediate_versions: list[DerivativeRelationVersion]
 
     def __post_init__(self) -> None:
-        if not self._intermediate_versions or not self._inital_version:
-            raise RelationWithoutLastVersionError
-
         for version, next_version in pairwise(self):
             if next(version.number) != next_version.number:
                 raise NotIncrementedRelationVersionError

src/tgdb/infrastructure/adapters/buffer.py

@@ -7,9 +7,9 @@
 from typing import Self
 
 from tgdb.application.common.ports.buffer import Buffer
-from tgdb.entities.transaction import TransactionCommit
+from tgdb.entities.horizon.transaction import PreparedCommit
 from tgdb.infrastructure.pydantic.commit_encoding import (
-    TransactionCommitListSchema,
+    PreparedCommitListSchema,
 )
 from tgdb.infrastructure.telethon.in_telegram_big_text import InTelegramBigText
 
@@ -47,8 +47,8 @@ def _refresh_overflow(self) -> None:
 
 
 @dataclass(frozen=True)
-class InTelegramReplicableTransactionCommitBuffer(Buffer[TransactionCommit]):
-    _buffer: Buffer[TransactionCommit]
+class InTelegramReplicablePreparedCommitBuffer(Buffer[PreparedCommit]):
+    _buffer: Buffer[PreparedCommit]
     _in_tg_encoded_commits: InTelegramBigText
 
     async def __aenter__(self) -> Self:
@@ -58,7 +58,7 @@ async def __aenter__(self) -> Self:
             return self
 
         commit_list_schema = (
-            TransactionCommitListSchema.model_validate_json(encoded_commits)
+            PreparedCommitListSchema.model_validate_json(encoded_commits)
         )
         for commit in commit_list_schema.commits:
             await self._buffer.add(commit)
@@ -72,12 +72,12 @@ async def __aexit__(
         traceback: TracebackType | None,
     ) -> None: ...
 
-    async def add(self, commit: TransactionCommit, /) -> None:
+    async def add(self, commit: PreparedCommit, /) -> None:
         await self._buffer.add(commit)
 
-    async def __aiter__(self) -> AsyncIterator[Sequence[TransactionCommit]]:
+    async def __aiter__(self) -> AsyncIterator[Sequence[PreparedCommit]]:
         async for commits in self._buffer:
-            commit_list_schema = TransactionCommitListSchema(commits=commits)
+            commit_list_schema = PreparedCommitListSchema(commits=commits)
             encoded_encoded_commits = commit_list_schema.model_dump_json()
 
             await self._in_tg_encoded_commits.set(encoded_encoded_commits)

src/tgdb/infrastructure/adapters/channel.py

@@ -0,0 +1,36 @@
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+
+from tgdb.application.common.ports.channel import Channel, Notification
+from tgdb.entities.horizon.horizon import (
+    InvalidTransactionStateError,
+    NoTransactionError,
+)
+from tgdb.entities.horizon.transaction import XID
+from tgdb.infrastructure.async_map import AsyncMap
+
+
+@dataclass(frozen=True)
+class AsyncMapChannel(Channel):
+    _async_map: AsyncMap[
+        XID, NoTransactionError | InvalidTransactionStateError | None
+    ]
+
+    async def publish(
+        self,
+        ok_commit_xids: Sequence[XID],
+        error_commit_map: Mapping[
+            XID, NoTransactionError | InvalidTransactionStateError
+        ],
+    ) -> None:
+        for ok_commit_xid in ok_commit_xids:
+            self._async_map[ok_commit_xid] = None
+            del self._async_map[ok_commit_xid]
+
+        for error_commit_xid, error in error_commit_map.items():
+            self._async_map[error_commit_xid] = error
+            del self._async_map[error_commit_xid]
+
+    async def wait(self, xid: XID) -> Notification:
+        notification_error = await self._async_map[xid]
+        return Notification(notification_error)

src/tgdb/infrastructure/adapters/clock.py

@@ -0,0 +1,28 @@
+from collections.abc import Generator
+from dataclasses import dataclass
+from time import perf_counter_ns
+from typing import Any
+
+from tgdb.application.common.ports.clock import Clock
+from tgdb.entities.time.logic_time import LogicTime
+
+
+@dataclass
+class InMemoryClock(Clock):
+    _time_counter: int = 0
+
+    def __await__(self) -> Generator[Any, Any, LogicTime]:
+        return self._time().__await__()
+
+    async def _time(self) -> LogicTime:
+        self._time_counter += 1
+        return self._time_counter
+
+
+@dataclass(frozen=True)
+class PerfCounterClock(Clock):
+    def __await__(self) -> Generator[Any, Any, LogicTime]:
+        return self._time().__await__()
+
+    async def _time(self) -> LogicTime:
+        return perf_counter_ns()

src/tgdb/infrastructure/adapters/logic_clock.py

@@ -0,0 +1,74 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+from in_memory_db import InMemoryDb
+
+from tgdb.application.common.ports.relations import (
+    NoRelationError,
+    NotUniqueRelationNumberError,
+    Relations,
+)
+from tgdb.entities.numeration.number import Number
+from tgdb.entities.relation.relation import Relation
+
+
+@dataclass(frozen=True)
+class InMemoryRelations(Relations):
+    _db: InMemoryDb[Relation]
+
+    async def relation(self, relation_number: Number) -> Relation:
+        """
+        :raises tgdb.application.common.ports.relations.NoRelationError:
+        """
+
+        relation = self._db.select_one(
+            lambda it: it.number() == relation_number
+        )
+
+        if relation is None:
+            raise NoRelationError(relation_number)
+
+    async def add(self, relation: Relation) -> None:
+        """
+        :raises tgdb.application.common.ports.relations.NotUniqueRelationNumberError:
+        """  # noqa: E501
+
+        selected_relation = self._db.select_one(
+            lambda it: it.number() == relation.number()
+        )
+
+        if selected_relation is not None:
+            raise NotUniqueRelationNumberError(relation.number())
+
+        self._db.insert(relation)
+
+
+@dataclass(frozen=True)
+class InMemoryRelations(Relations):
+    _db: InMemoryDb[Relation]
+
+    async def relation(self, relation_number: Number) -> Relation:
+        """
+        :raises tgdb.application.common.ports.relations.NoRelationError:
+        """
+
+        relation = self._db.select_one(
+            lambda it: it.number() == relation_number
+        )
+
+        if relation is None:
+            raise NoRelationError(relation_number)
+
+    async def add(self, relation: Relation) -> None:
+        """
+        :raises tgdb.application.common.ports.relations.NotUniqueRelationNumberError:
+        """  # noqa: E501
+
+        selected_relation = self._db.select_one(
+            lambda it: it.number() == relation.number()
+        )
+
+        if selected_relation is not None:
+            raise NotUniqueRelationNumberError(relation.number())
+
+        self._db.insert(relation)

src/tgdb/infrastructure/adapters/relations.py

@@ -0,0 +1,19 @@
+from types import TracebackType
+
+from tgdb.application.common.ports.shared_horizon import SharedHorizon
+from tgdb.entities.horizon.horizon import Horizon
+
+
+class InMemorySharedHorizon(SharedHorizon):
+    _horizon: Horizon
+
+    async def __aenter__(self) -> Horizon:
+        return self._horizon
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+        /,
+    ) -> None: ...

src/tgdb/infrastructure/adapters/shared_horizon.py

@@ -4,8 +4,7 @@
 
 from in_memory_db import InMemoryDb
 
-from tgdb.application.common.ports.heap import Heap
-from tgdb.entities.message import Message
+from tgdb.application.common.ports.tuples import Tuples
 from tgdb.entities.row import (
     DeletedRow,
     MutatedRow,
@@ -20,7 +19,7 @@
 
 
 @dataclass(frozen=True, unsafe_hash=False)
-class InMemoryHeap(Heap):
+class InMemoryTuples(Tuples):
     _db: InMemoryDb[Row]
 
     async def map(self, effects: Sequence[TransactionEffect]) -> None:

src/tgdb/infrastructure/adapters/heap.py …c/tgdb/infrastructure/adapters/tuples.pysrc/tgdb/infrastructure/adapters/heap.py renamed to src/tgdb/infrastructure/adapters/tuples.py src/tgdb/infrastructure/adapters/heap.py renamed to src/tgdb/infrastructure/adapters/tuples.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass, field
+from uuid import UUID, uuid4
+
+from tgdb.application.common.ports.uuids import UUIDs
+
+
+@dataclass(frozen=True)
+class UUIDs4(UUIDs):
+    async def random_uuid(self) -> UUID:
+        return uuid4()
+
+
+@dataclass
+class MonotonicUUIDs(UUIDs):
+    _counter: int = field(init=False, default=0)
+
+    async def random_uuid(self) -> UUID:
+        next_uuid = UUID(int=self._counter)
+        self._counter += 1
+
+        return next_uuid