Validation Middleware

validation_middleware

Middleware that validates requests before delegation and short-circuits on failure.

ValidationMiddleware

Bases: Middleware[RequestType, ResponseType]

Validates each request before forwarding to the downstream handler.

Delegates all validation logic to a caller-supplied validator
callable. When the validator returns a response, the middleware
short-circuits the pipeline — the downstream handler is never
called. When the validator returns None, the request is
considered valid and passes through unchanged.

Responsibilities

• Invoke the validator with every incoming request.
• Short-circuit and return the validator's response when
  it is not None.
• Pass the request through unchanged when the validator
  returns None.

Non-Responsibilities

• Define validation rules — those live in the caller.
• Format error responses — the validator returns the
  final ResponseType directly.
• Handle errors raised by downstream middleware or the handler.

Example

def validate(request: MyRequest) -> MyResponse | None:
    if request.is_invalid:
        return MyResponse(error="Invalid request")
    return None  # pass-through to next handler


mw = ValidationMiddleware[MyRequest, MyResponse](validator=validate)
response = await mw.process(request, next_handler)

Source code in src/forging_blocks/presentation/builtin/validation_middleware.py

class ValidationMiddleware[RequestType, ResponseType](Middleware[RequestType, ResponseType]):
    """Validates each request before forwarding to the downstream handler.

    Delegates all validation logic to a caller-supplied ``validator``
    callable.  When the validator returns a response, the middleware
    short-circuits the pipeline — the downstream handler is never
    called.  When the validator returns ``None``, the request is
    considered valid and passes through unchanged.

    Responsibilities:
        - Invoke the validator with every incoming request.
        - Short-circuit and return the validator's response when
          it is not ``None``.
        - Pass the request through unchanged when the validator
          returns ``None``.

    Non-Responsibilities:
        - Define validation rules — those live in the caller.
        - Format error responses — the validator returns the
          final ``ResponseType`` directly.
        - Handle errors raised by downstream middleware or the handler.

    Example:
        ```python
        def validate(request: MyRequest) -> MyResponse | None:
            if request.is_invalid:
                return MyResponse(error="Invalid request")
            return None  # pass-through to next handler


        mw = ValidationMiddleware[MyRequest, MyResponse](validator=validate)
        response = await mw.process(request, next_handler)
        ```

    """

    __slots__ = ("_validator",)

    def __init__(self, validator: Callable[[RequestType], ResponseType | None]) -> None:
        """Wrap *validator* so every request is validated.

        Args:
            validator: A callable that receives the request and returns
                either a ``ResponseType`` (to short-circuit) or ``None``
                (to pass through).

        """
        self._validator = validator

    async def process(
        self,
        request: RequestType,
        next_handler: NextHandler[RequestType, ResponseType],
    ) -> ResponseType:
        """Validate the request, short-circuiting on failure.

        Args:
            request: The incoming request.
            next_handler: The next callable in the pipeline chain.

        Returns:
            - The validator's response when it short-circuits.
            - The downstream handler's response otherwise.

        """
        error_response = self._validator(request)
        if error_response is not None:
            return error_response
        return await next_handler(request)

__init__(validator: Callable[[RequestType], ResponseType | None]) -> None

Wrap validator so every request is validated.

Parameters:

Name	Type	Description	Default
validator	Callable[[RequestType], ResponseType | None]	A callable that receives the request and returns either a ResponseType (to short-circuit) or None (to pass through).	required

Source code in src/forging_blocks/presentation/builtin/validation_middleware.py

def __init__(self, validator: Callable[[RequestType], ResponseType | None]) -> None:
    """Wrap *validator* so every request is validated.

    Args:
        validator: A callable that receives the request and returns
            either a ``ResponseType`` (to short-circuit) or ``None``
            (to pass through).

    """
    self._validator = validator

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

Validate the request, short-circuiting on failure.

Parameters:

Name	Type	Description	Default
request	RequestType	The incoming request.	required
next_handler	NextHandler[RequestType, ResponseType]	The next callable in the pipeline chain.	required

Returns:

Type	Description
ResponseType	The validator's response when it short-circuits.
ResponseType	The downstream handler's response otherwise.

Source code in src/forging_blocks/presentation/builtin/validation_middleware.py

async def process(
    self,
    request: RequestType,
    next_handler: NextHandler[RequestType, ResponseType],
) -> ResponseType:
    """Validate the request, short-circuiting on failure.

    Args:
        request: The incoming request.
        next_handler: The next callable in the pipeline chain.

    Returns:
        - The validator's response when it short-circuits.
        - The downstream handler's response otherwise.

    """
    error_response = self._validator(request)
    if error_response is not None:
        return error_response
    return await next_handler(request)