Skip to content

Presentation Adapter

presentation_adapter

Orchestrator that wires transport adapters to a use case with error handling.

PresentationAdapter handles both returned Err and raised
Exception / Error, so callers may choose their error-signalling
style without changing the adapter.

PresentationAdapter

Orchestrates the full request/response lifecycle for a use case.

Usage::

adapter = PresentationAdapter(
    use_case=create_order_use_case,
    request_adapter=JsonRequestAdapter(),
    response_adapter=JsonResponseAdapter(),
    error_presenter=ErrorPresenter(),
)
response = await adapter.handle(http_request)
Source code in src/forging_blocks/presentation/presentation_adapter.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
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class PresentationAdapter[RawRequest, UseCaseInput, UseCaseOutput, RawResponse]:
    """Orchestrates the full request/response lifecycle for a use case.

    Usage::

        adapter = PresentationAdapter(
            use_case=create_order_use_case,
            request_adapter=JsonRequestAdapter(),
            response_adapter=JsonResponseAdapter(),
            error_presenter=ErrorPresenter(),
        )
        response = await adapter.handle(http_request)
    """

    __slots__ = (
        "_error_presenter",
        "_request_adapter",
        "_response_adapter",
        "_status_mapper",
        "_use_case",
    )

    def __init__(
        self,
        use_case: "UseCase[UseCaseInput, UseCaseOutput]",
        request_adapter: RequestAdapter[RawRequest, UseCaseInput],
        response_adapter: ResponseAdapter[UseCaseOutput, RawResponse],
        error_presenter: ErrorPresenter | None = None,
    ) -> None:
        """Wire the adapter with its collaborators.

        Args:
            use_case: The application use case to invoke.
            request_adapter: Translates transport requests into
                use-case input.
            response_adapter: Translates use-case output into
                transport responses (success and error).
            error_presenter: Optional error formatter. When omitted,
                exceptions propagate unchanged.
        """
        self._use_case = use_case
        self._request_adapter = request_adapter
        self._response_adapter = response_adapter
        self._error_presenter = error_presenter
        self._status_mapper = ErrorStatusCodeMapper()

    async def handle(self, raw_request: RawRequest) -> RawResponse:
        """Process *raw_request* through the full lifecycle.

        Handles both ``Result.Err`` (if the use case returns a
        ``Result``) and raised exceptions so callers may choose
        their preferred error style.

        Args:
            raw_request: The transport-level request.

        Returns:
            A transport-level response (success or error, depending
            on the outcome).

        Raises:
            Exception: When *error_presenter* is ``None`` and the
                use case raises, the original exception propagates.
        """
        try:
            use_case_input = self._request_adapter.adapt(raw_request)
            use_case_output = await self._use_case.execute(use_case_input)
        except Exception as exc:
            if self._error_presenter is None:
                raise
            view_model = self._error_presenter.to_view_model(exc)
            mapped = self._status_mapper.map(view_model)
            return self._response_adapter.adapt_error(mapped)

        if isinstance(use_case_output, Result):
            result = cast("Result[UseCaseOutput, object]", use_case_output)
            if result.is_err:
                if self._error_presenter is None:
                    error_value = result.error
                    if isinstance(error_value, BaseException):
                        raise error_value
                    raise RuntimeError(str(error_value))
                view_model = self._error_presenter.to_view_model(result.error)
                mapped = self._status_mapper.map(view_model)
                return self._response_adapter.adapt_error(mapped)
            return self._response_adapter.adapt(result.value)

        return self._response_adapter.adapt(use_case_output)

__init__(use_case: UseCase[UseCaseInput, UseCaseOutput], request_adapter: RequestAdapter[RawRequest, UseCaseInput], response_adapter: ResponseAdapter[UseCaseOutput, RawResponse], error_presenter: ErrorPresenter | None = None) -> None

Wire the adapter with its collaborators.

Parameters:

Name Type Description Default
use_case UseCase[UseCaseInput, UseCaseOutput]

The application use case to invoke.

 required
request_adapter RequestAdapter[RawRequest, UseCaseInput]

Translates transport requests into
use-case input.

 required
response_adapter ResponseAdapter[UseCaseOutput, RawResponse]

Translates use-case output into
transport responses (success and error).

 required
error_presenter ErrorPresenter | None

Optional error formatter. When omitted,
exceptions propagate unchanged.

 None
Source code in src/forging_blocks/presentation/presentation_adapter.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(
    self,
    use_case: "UseCase[UseCaseInput, UseCaseOutput]",
    request_adapter: RequestAdapter[RawRequest, UseCaseInput],
    response_adapter: ResponseAdapter[UseCaseOutput, RawResponse],
    error_presenter: ErrorPresenter | None = None,
) -> None:
    """Wire the adapter with its collaborators.

    Args:
        use_case: The application use case to invoke.
        request_adapter: Translates transport requests into
            use-case input.
        response_adapter: Translates use-case output into
            transport responses (success and error).
        error_presenter: Optional error formatter. When omitted,
            exceptions propagate unchanged.
    """
    self._use_case = use_case
    self._request_adapter = request_adapter
    self._response_adapter = response_adapter
    self._error_presenter = error_presenter
    self._status_mapper = ErrorStatusCodeMapper()

handle(raw_request: RawRequest) -> RawResponse async

Process raw_request through the full lifecycle.

Handles both Result.Err (if the use case returns a
Result) and raised exceptions so callers may choose
their preferred error style.

Parameters:

Name Type Description Default
raw_request RawRequest

The transport-level request.

 required

Returns:

Type Description
RawResponse

A transport-level response (success or error, depending
RawResponse

on the outcome).

Raises:

Type Description
Exception

When error_presenter is None and the
use case raises, the original exception propagates.
Source code in src/forging_blocks/presentation/presentation_adapter.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def handle(self, raw_request: RawRequest) -> RawResponse:
    """Process *raw_request* through the full lifecycle.

    Handles both ``Result.Err`` (if the use case returns a
    ``Result``) and raised exceptions so callers may choose
    their preferred error style.

    Args:
        raw_request: The transport-level request.

    Returns:
        A transport-level response (success or error, depending
        on the outcome).

    Raises:
        Exception: When *error_presenter* is ``None`` and the
            use case raises, the original exception propagates.
    """
    try:
        use_case_input = self._request_adapter.adapt(raw_request)
        use_case_output = await self._use_case.execute(use_case_input)
    except Exception as exc:
        if self._error_presenter is None:
            raise
        view_model = self._error_presenter.to_view_model(exc)
        mapped = self._status_mapper.map(view_model)
        return self._response_adapter.adapt_error(mapped)

    if isinstance(use_case_output, Result):
        result = cast("Result[UseCaseOutput, object]", use_case_output)
        if result.is_err:
            if self._error_presenter is None:
                error_value = result.error
                if isinstance(error_value, BaseException):
                    raise error_value
                raise RuntimeError(str(error_value))
            view_model = self._error_presenter.to_view_model(result.error)
            mapped = self._status_mapper.map(view_model)
            return self._response_adapter.adapt_error(mapped)
        return self._response_adapter.adapt(result.value)

    return self._response_adapter.adapt(use_case_output)