Middleware

middleware

Protocol for middleware interceptors in the presentation layer.

Middleware defines the contract for cross-cutting interceptors that
sit between the adapter and the application handler. Each middleware
may inspect, transform, or short-circuit the request before delegating
to the next handler in the chain.

Middleware

Bases: Protocol

Structural protocol for a middleware interceptor.

Implementations must not depend on infrastructure details.
Middleware is exclusively a composition primitive — it does not
extend Port because its shape (request, next_handler) -> response differs from the single-input, single-output port
contract.

Short-circuiting is supported: a middleware may return a response
without calling next_handler, skipping all downstream middleware
and the terminal handler.

Source code in src/forging_blocks/presentation/middleware.py

@runtime_checkable
class Middleware[RequestType, ResponseType](Protocol):
    """Structural protocol for a middleware interceptor.

    Implementations must not depend on infrastructure details.
    Middleware is exclusively a composition primitive — it does not
    extend ``Port`` because its shape ``(request, next_handler) ->
    response`` differs from the single-input, single-output port
    contract.

    Short-circuiting is supported: a middleware may return a response
    without calling *next_handler*, skipping all downstream middleware
    and the terminal handler.
    """

    async def process(
        self,
        request: RequestType,
        next_handler: NextHandler[RequestType, ResponseType],
    ) -> ResponseType:
        """Intercept *request* and optionally delegate to *next_handler*.

        Args:
            request: The incoming request to process.
            next_handler: The next callable in the pipeline chain.
                Must be called to continue processing. If not called,
                the chain short-circuits and this method's return
                value becomes the final response.

        Returns:
            The response after processing.
        """
        ...

process(request: RequestType, next_handler: NextHandler[RequestType, ResponseType]) -> ResponseType async

Intercept request and optionally delegate to next_handler.

Parameters:

Name	Type	Description	Default
request	RequestType	The incoming request to process.	required
next_handler	NextHandler[RequestType, ResponseType]	The next callable in the pipeline chain. Must be called to continue processing. If not called, the chain short-circuits and this method's return value becomes the final response.	required

Returns:

Type	Description
ResponseType	The response after processing.

Source code in src/forging_blocks/presentation/middleware.py

async def process(
    self,
    request: RequestType,
    next_handler: NextHandler[RequestType, ResponseType],
) -> ResponseType:
    """Intercept *request* and optionally delegate to *next_handler*.

    Args:
        request: The incoming request to process.
        next_handler: The next callable in the pipeline chain.
            Must be called to continue processing. If not called,
            the chain short-circuits and this method's return
            value becomes the final response.

    Returns:
        The response after processing.
    """
    ...