Skip to content

Transaction Manager Port

transaction_manager_port

Outbound port for transactional boundaries.

Defines the TransactionManagerPort contract for explicit transaction
control (begin, commit, rollback) and transactional function execution.

Responsibilities
  • Begin, commit, and roll back transactions.
  • Execute arbitrary functions within transactional boundaries.
Non-Responsibilities
  • Implement business logic.
  • Manage connection pooling or resource cleanup.

TransactionManagerPort

Bases: OutboundPort

Outbound port for explicit transaction management.

Class Type Parameters:

Name Bound or Constraints Description Default
TransactionSessionContext

The application-defined session context for transactions.

 required
TransactionErrorType

The error type for transactional failures.

 required
Responsibilities
  • Begin, commit, and roll back transactions.
  • Execute arbitrary functions within transactional boundaries.
Non-Responsibilities
  • Implement business logic.
  • Manage connection pooling or resource cleanup.
Example 
tm = MyTransactionManager[DbSession, TransactionError]()
await tm.begin(session)
await tm.commit()
Source code in src/forging_blocks/application/ports/outbound/transaction_manager_port.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class TransactionManagerPort[TransactionSessionContext, TransactionErrorType](OutboundPort):
    """Outbound port for explicit transaction management.

    Type Args:
        TransactionSessionContext: The application-defined session context for transactions.
        TransactionErrorType: The error type for transactional failures.

    Responsibilities:
        - Begin, commit, and roll back transactions.
        - Execute arbitrary functions within transactional boundaries.

    Non-Responsibilities:
        - Implement business logic.
        - Manage connection pooling or resource cleanup.

    Example:
        ```python
        tm = MyTransactionManager[DbSession, TransactionError]()
        await tm.begin(session)
        await tm.commit()
        ```
    """

    @abstractmethod
    async def begin(self, context: TransactionSessionContext) -> 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: TransactionSessionContext,
        *args: object,
        **kwargs: object,
    ) -> Result[ResponseType, TransactionErrorType]:
        """Execute *fn* inside a begin/commit boundary."""
        ...

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

Start a new transaction.

Source code in src/forging_blocks/application/ports/outbound/transaction_manager_port.py 
45
46
47
48
@abstractmethod
async def begin(self, context: TransactionSessionContext) -> 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 
50
51
52
53
@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 
55
56
57
58
@abstractmethod
async def rollback(self) -> Result[None, TransactionErrorType]:
    """Roll back the current transaction."""
    ...

execute_in_transaction(fn: Callable[..., Awaitable[ResponseType]], context: TransactionSessionContext, *args: object, **kwargs: object) -> 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 
60
61
62
63
64
65
66
67
68
69
@abstractmethod
async def execute_in_transaction[ResponseType](
    self,
    fn: Callable[..., Awaitable[ResponseType]],
    context: TransactionSessionContext,
    *args: object,
    **kwargs: object,
) -> Result[ResponseType, TransactionErrorType]:
    """Execute *fn* inside a begin/commit boundary."""
    ...