feat: add a part of schema versioning

Author: emptybutton

src/tgdb/application/errors/__init__.py

@@ -0,0 +1,7 @@
+class NotActiveNodeError(Exception): ...
+
+
+class ActiveNodeError(Exception): ...
+
+
+class InvalidInputOperatorError(Exception): ...

src/tgdb/application/errors/common.py

@@ -1,13 +1,11 @@
 from dataclasses import dataclass
 
+from tgdb.application.errors.common import InvalidInputOperatorError
 from tgdb.application.ports.operator_serialization import OperatorSerialization
 from tgdb.application.ports.queque import Queque
 from tgdb.entities.operator import Operator
 
 
-class InputOperatorError(Exception): ...
-
-
 @dataclass(frozen=True)
 class InputOperator[SerializedOperatorsT]:
     input_operators: Queque[Operator]
@@ -17,14 +15,14 @@ async def __call__(
         self, serialized_operator: SerializedOperatorsT
     ) -> None:
         """
-        :raises tgdb.application.input_operator.InputOperatorError:
+        :raises tgdb.application.errors.common.InvalidInputOperatorError:
         """
 
         input_operator = await self.operator_serialization.deserialized(
             serialized_operator
         )
 
         if input_operator is None:
-            raise InputOperatorError
+            raise InvalidInputOperatorError
 
         await self.input_operators.push(input_operator)

src/tgdb/application/input_operator.py

@@ -5,6 +5,9 @@
 
 
 class Heap(ABC):
+    @abstractmethod
+    async def rows(self, schema: str, )
+
     @abstractmethod
     async def map(self, effects: Sequence[TransactionEffect], /) -> None: ...
 

src/tgdb/application/ports/heap.py

@@ -0,0 +1,11 @@
+from abc import ABC, abstractmethod
+from contextlib import AbstractAsyncContextManager
+
+from tgdb.entities.transaction_horizon import TransactionHorizon
+
+
+class SharedHorizon(
+    AbstractAsyncContextManager[TransactionHorizon | None], ABC
+):
+    @abstractmethod
+    async def set(self, horizon: TransactionHorizon | None) -> None: ...

src/tgdb/application/ports/shared_horizon.py

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+
+from tgdb.application.errors.common import NotActiveNodeError
+from tgdb.application.ports.shared_horizon import SharedHorizon
+
+
+@dataclass(frozen=True)
+class StartDown:
+    shared_horizon: SharedHorizon
+
+    async def __call__(self) -> None:
+        """
+        :raises tgdb.application.errors.common.NotActiveNodeError:
+        """
+
+        async with self.shared_horizon as horizon:
+            if horizon is not None:
+                raise NotActiveNodeError
+
+            await self.shared_horizon.set(None)

src/tgdb/application/start_down.py

@@ -0,0 +1,30 @@
+from dataclasses import dataclass
+
+from tgdb.application.errors.common import ActiveNodeError
+from tgdb.application.ports.shared_horizon import SharedHorizon
+from tgdb.entities.logic_time import LogicTime
+from tgdb.entities.transaction_horizon import create_transaction_horizon
+
+
+@dataclass(frozen=True)
+class StartUp:
+    shared_horizon: SharedHorizon
+
+    async def __call__(
+        self,
+        horizon_max_width: LogicTime | None,
+        horizon_max_height: int | None,
+    ) -> None:
+        """
+        :raises tgdb.application.errors.common.ActiveNodeError:
+        """
+
+        async with self.shared_horizon as horizon:
+            if horizon is not None:
+                raise ActiveNodeError
+
+            new_horizon = create_transaction_horizon(
+                horizon_max_width, horizon_max_height
+            )
+
+            await self.shared_horizon.set(new_horizon)

src/tgdb/application/start_up.py

@@ -1,3 +1,7 @@
-def not_none[ValueT](value: ValueT | None) -> ValueT:
-    assert value is not None
-    return value
+def not_none[ValueT](
+    value: ValueT | None, *, error: Exception | type[Exception] = ValueError
+) -> ValueT:
+    if value is not None:
+        return value
+
+    raise error

src/tgdb/entities/assert_.py

@@ -4,29 +4,46 @@
 
 from tgdb.entities.logic_time import LogicTime
 from tgdb.entities.mark import Mark
-from tgdb.entities.row import RowEffect
-
-
-type IntermediateOperator = RowEffect | Mark
+from tgdb.entities.row import DeletedRow, MutatedRow, NewRow, RowEffect
+from tgdb.entities.transaction import TransactionIsolation
 
 
 @dataclass(frozen=True)
 class StartOperator:
     transaction_id: UUID
+    transaction_isolation: TransactionIsolation
+
+
+type IntermediateOperatorEffect = RowEffect | Mark
 
 
 @dataclass(frozen=True)
-class CommitOperator:
+class IntermediateOperator[
+    EffectT: IntermediateOperatorEffect = IntermediateOperatorEffect
+]:
     transaction_id: UUID
-    operators: Sequence[IntermediateOperator]
+    effect: IntermediateOperatorEffect
 
 
 @dataclass(frozen=True)
 class RollbackOperator:
     transaction_id: UUID
 
 
-type Operator = StartOperator | CommitOperator | RollbackOperator
+@dataclass(frozen=True)
+class CommitOperator:
+    transaction_id: UUID
+    operators: Sequence[
+        IntermediateOperator[NewRow | MutatedRow | DeletedRow | Mark]
+    ]
+
+
+type Operator = (
+    StartOperator
+    | IntermediateOperator
+    | RollbackOperator
+    | CommitOperator
+)
 
 
 @dataclass(frozen=True)