chore(infrastructure): add not completed adapters

Author: emptybutton

src/tgdb/application/ports/heap.py

@@ -18,7 +18,7 @@ async def row(
     ) -> Row | None: ...
 
     @abstractmethod
-    async def rows(
+    def rows(
         self,
         schema: RowSchema,
         attribute_number: int,

src/tgdb/infrastructure/adapters/__init__.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+from tgdb.application.ports.commit_serialization import CommitSerialization
+from tgdb.entities.transaction import TransactionCommit
+
+
+@dataclass(frozen=True)
+class CommitSerializationToCommit(CommitSerialization[TransactionCommit]):
+    async def serialized(self, commit: TransactionCommit) -> TransactionCommit:
+        return commit

src/tgdb/infrastructure/adapters/commit_serialization.py

@@ -0,0 +1,77 @@
+from collections.abc import AsyncIterator
+from dataclasses import dataclass, field
+from functools import partial
+
+from tgdb.application.ports.heap import Heap
+from tgdb.entities.row import Row, RowAttribute, RowSchema
+from tgdb.entities.transaction import TransactionEffect
+
+
+@dataclass(frozen=True, unsafe_hash=False)
+class InMemoryHeap(Heap):
+    _row_by_row_id: dict[RowAttribute, Row] = field(default_factory=dict)
+
+    async def insert(self, row: Row) -> None:
+        if row.id in self._row_by_row_id:
+            raise ValueError
+
+        self._row_by_row_id[row.id] = row
+
+    async def row(
+        self,
+        schema: RowSchema,
+        attribute_number: int,
+        attribute: RowAttribute | None = None,
+    ) -> Row | None:
+        for row in self._row_by_row_id.values():
+            if self._is_valid(schema, attribute_number, attribute, row):
+                return row
+
+        return None
+
+    async def rows(
+        self,
+        schema: RowSchema,
+        attribute_number: int,
+        attribute: RowAttribute | None = None,
+    ) -> AsyncIterator[Row]:
+        is_valid = partial(self._is_valid, schema, attribute_number, attribute)
+        rows = tuple(filter(is_valid, self._row_by_row_id.values()))
+
+        for row in rows:
+            yield row
+
+    async def update(self, row: Row) -> None:
+        if row.id not in self._row_by_row_id:
+            raise ValueError
+
+        self._row_by_row_id[row.id] = row
+
+    async def delete(self, row: Row) -> None:
+        del self._row_by_row_id[row.id]
+
+    async def map(self, effect: TransactionEffect) -> None:
+        for new_row in effect.new_values:
+            await self.insert(new_row)
+
+        for mutated_row in effect.mutated_values:
+            await self.update(mutated_row)
+
+        for dead_row in effect.dead_values:
+            await self.delete(dead_row)
+
+    def _is_valid(
+        self,
+        schema: RowSchema,
+        attribute_number: int,
+        attribute: RowAttribute | None,
+        row: Row,
+    ) -> bool:
+        return (
+            row.schema == schema
+            and len(row) >= attribute_number + 1
+            and (
+                attribute is not None
+                or row[attribute_number] == attribute
+            )
+        )

src/tgdb/infrastructure/adapters/heap.py

@@ -0,0 +1,35 @@
+from abc import abstractmethod
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+
+from tgdb.application.ports.log import LogSlot
+from tgdb.entities.logic_time import LogicTime
+from tgdb.entities.operator import AppliedOperator
+from tgdb.infrastructure.async_queque import AsyncQueque
+
+
+@dataclass(frozen=True, unsafe_hash=False)
+class AsyncQuequeLogSlot(LogSlot):
+    _queque: AsyncQueque[AppliedOperator]
+
+    async def push(self, operator: AppliedOperator, /) -> None:
+        if self._queque and operator.time <= self._queque[-1].time:
+            raise ValueError
+
+        self._queque.push(operator)
+
+    async def __call__(self, *, block: bool) -> AsyncIterator[AppliedOperator]:
+        if block:
+            async for operator in self._queque:
+                yield operator
+
+            return
+
+        for operator in self._queque:
+            yield operator
+
+    @abstractmethod
+    async def commit(self, offset: LogicTime) -> None: ...
+
+    @abstractmethod
+    async def offset(self) -> LogicTime | None: ...

src/tgdb/infrastructure/adapters/log.py

@@ -0,0 +1,12 @@
+from collections.abc import Generator
+from dataclasses import dataclass
+
+
+@dataclass
+class InMemoryLogicClock:
+    _current_time: int = 0
+
+    def __await__(self) -> Generator[None, None, int]:
+        self._current_time += 1
+        yield
+        return self._current_time

src/tgdb/infrastructure/adapters/logic_clock.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+from tgdb.application.ports.operator_serialization import OperatorSerialization
+from tgdb.entities.operator import Operator
+
+
+@dataclass(frozen=True)
+class OperatorSerializationToOperator(OperatorSerialization[Operator | None]):
+    async def deserialized(self, operator: Operator | None) -> Operator | None:
+        return operator

src/tgdb/infrastructure/adapters/operator_serialization.py

@@ -0,0 +1,42 @@
+from abc import ABC, abstractmethod
+from asyncio import Event, Semaphore
+from collections.abc import AsyncIterator
+from dataclasses import dataclass, field
+
+from tgdb.application.ports.queque import Queque
+from tgdb.infrastructure.async_queque import AsyncQueque
+
+
+@dataclass
+class InMemoryQueque[ValueT](Queque[ValueT]):
+    _queque: AsyncQueque[ValueT]
+    _sync_reading_completed: Event = field(init=False, default_factory=Event)
+
+    async def async_push(self, value: ValueT) -> None:
+        self._queque.push(value)
+
+    async def sync_push(self, value: ValueT) -> None:
+        self._sync_reading_completed.clear()
+        self._queque.push(value)
+        await self._sync_reading_completed.wait()
+
+    async def async_(self) -> AsyncIterator[ValueT]:
+        values = self._async_values()
+        first_value = await anext(values)
+
+        return self._values(first_value, values)
+
+    async def _values(
+        self, first_value: ValueT, values: AsyncIterator[ValueT]
+    ) -> AsyncIterator[ValueT]:
+        yield first_value
+
+        async for value in values:
+            yield value
+
+    async def _async_values(self) -> AsyncIterator[ValueT]:
+        async for value in self._queque:
+            yield value
+
+    async def sync(self) -> AsyncIterator[ValueT]:
+        for value in 

src/tgdb/infrastructure/adapters/queque.py

@@ -1,11 +1,18 @@
 from asyncio import Event
 from collections import deque
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Iterator, Sequence
 from dataclasses import dataclass, field
+from typing import Any, overload
+
+
+#    |
+#   a
+#    |
+#     |
 
 
 @dataclass(frozen=True, unsafe_hash=False)
-class AsyncQueque[ValueT]:
+class AsyncQueque[ValueT](Sequence[ValueT]):
     _values: deque[ValueT] = field(default_factory=deque)
     _offset_by_event: dict[Event, int] = field(
         default_factory=dict, init=False
@@ -14,6 +21,25 @@ class AsyncQueque[ValueT]:
     def __len__(self) -> int:
         return len(self._values)
 
+    def __bool__(self) -> bool:
+        return bool(self._values)
+
+    def __iter__(self) -> Iterator[ValueT]:
+        return iter(self._values)
+
+    @overload
+    def __getitem__(self, index: int, /) -> ValueT: ...
+
+    @overload
+    def __getitem__(
+        self, slice_: "slice[Any, Any, Any]", /
+    ) -> Sequence[ValueT]: ...
+
+    def __getitem__(
+        self, value: "int | slice[Any, Any, Any]", /
+    ) -> Sequence[ValueT] | ValueT:
+        return self._values[value]
+
     def push(self, value: ValueT) -> None:
         self._values.append(value)
 
@@ -22,6 +48,10 @@ def push(self, value: ValueT) -> None:
 
     async def __aiter__(self) -> AsyncIterator[ValueT]:
         event = Event()
+
+        if self._values:
+            event.set()
+
         self._offset_by_event[event] = -1
 
         while True:
@@ -34,7 +64,10 @@ async def __aiter__(self) -> AsyncIterator[ValueT]:
 
             yield new_value
 
-    def _refresh(self) -> None:
+            if self._offset_by_event[event] == len(self._values) - 1:
+                event.clear()
+
+    def refresh(self) -> None:
         min_offset = min(self._offset_by_event.values())
 
         if min_offset >= 0:

src/tgdb/infrastructure/async_queque.py

@@ -20,9 +20,6 @@ async def set(self, value: ValueT, /) -> None:
             self._encoded(value),
         )
 
-    def __await__(self) -> Generator[None, None, ValueT]:
-        return self.get().__await__()
-
     async def get(self) -> ValueT:
         pointer_message = await self._pointer_message()