Skip to content

Transaction Manager Port

transaction_manager_port

Outbound port for transactional boundaries.

TransactionManagerPort

Bases: ABC

Abstract transaction manager for the application layer.

Source code in src/forging_blocks/application/ports/outbound/transaction_manager_port.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
class TransactionManagerPort[TransactionErrorType](ABC):
    """Abstract transaction manager for the application layer."""

    @abstractmethod
    async def begin(self, context: TransactionContext) -> Result[None, TransactionErrorType]:
        """Start a new transaction."""
        ...

    @abstractmethod
    async def commit(self) -> Result[None, TransactionErrorType]:
        """Commit the current transaction."""
        ...

    @abstractmethod
    async def rollback(self) -> Result[None, TransactionErrorType]:
        """Roll back the current transaction."""
        ...

    @abstractmethod
    async def execute_in_transaction[ResponseType](
        self,
        fn: Callable[..., Awaitable[ResponseType]],
        context: TransactionContext,
        *args: Any,
        **kwargs: Any,
    ) -> Result[ResponseType, TransactionErrorType]:
        """Execute *fn* inside a begin/commit boundary."""
        ...

begin(context: TransactionContext) -> Result[None, TransactionErrorType] abstractmethod async

Start a new transaction.

Source code in src/forging_blocks/application/ports/outbound/transaction_manager_port.py 
14
15
16
17
@abstractmethod
async def begin(self, context: TransactionContext) -> Result[None, TransactionErrorType]:
    """Start a new transaction."""
    ...

commit() -> Result[None, TransactionErrorType] abstractmethod async

Commit the current transaction.

Source code in src/forging_blocks/application/ports/outbound/transaction_manager_port.py 
19
20
21
22
@abstractmethod
async def commit(self) -> Result[None, TransactionErrorType]:
    """Commit the current transaction."""
    ...

rollback() -> Result[None, TransactionErrorType] abstractmethod async

Roll back the current transaction.

Source code in src/forging_blocks/application/ports/outbound/transaction_manager_port.py 
24
25
26
27
@abstractmethod
async def rollback(self) -> Result[None, TransactionErrorType]:
    """Roll back the current transaction."""
    ...

execute_in_transaction(fn: Callable[..., Awaitable[ResponseType]], context: TransactionContext, *args: Any, **kwargs: Any) -> Result[ResponseType, TransactionErrorType] abstractmethod async

Execute fn inside a begin/commit boundary.

Source code in src/forging_blocks/application/ports/outbound/transaction_manager_port.py 
29
30
31
32
33
34
35
36
37
38
@abstractmethod
async def execute_in_transaction[ResponseType](
    self,
    fn: Callable[..., Awaitable[ResponseType]],
    context: TransactionContext,
    *args: Any,
    **kwargs: Any,
) -> Result[ResponseType, TransactionErrorType]:
    """Execute *fn* inside a begin/commit boundary."""
    ...