Middleware API

The middleware module defines the middleware system for Plinx applications. Middleware components process requests before they reach handlers and responses before they're returned to clients.

Middleware Class

plinx.middleware.Middleware

Base class for all Plinx middleware components.

The middleware system in Plinx follows a nested pattern where each middleware wraps the application or another middleware component. This allows for a chain of processing both before a request reaches the application and after the application generates a response.

Middleware classes should inherit from this base class and override the process_request and process_response methods to implement custom behavior.

The middleware execution flow works like this: 1. Client request comes in 2. Each middleware's process_request is called from outermost to innermost 3. The application handles the request 4. Each middleware's process_response is called from innermost to outermost 5. The response is sent back to the client

Examples:

class LoggingMiddleware(Middleware):
    def process_request(self, request):
        print(f"Request: {request.path}")

    def process_response(self, request, response):
        print(f"Response: {response.status_code}")

app = Plinx()
app.add_middleware(LoggingMiddleware)

Middleware that modifies the request or response:

class AuthMiddleware(Middleware):
    def process_request(self, request):
        request.user = None
        auth_header = request.headers.get("Authorization", "")
        if auth_header.startswith("Bearer "):
            token = auth_header[7:]
            request.user = self.get_user_from_token(token)

    def get_user_from_token(self, token):
        # Implementation to validate token and return user
        pass

Source code in plinx/middleware.py

class Middleware:
    """
    Base class for all Plinx middleware components.

    The middleware system in Plinx follows a nested pattern where each middleware
    wraps the application or another middleware component. This allows for a chain
    of processing both before a request reaches the application and after the
    application generates a response.

    Middleware classes should inherit from this base class and override the
    `process_request` and `process_response` methods to implement custom behavior.

    The middleware execution flow works like this:
    1. Client request comes in
    2. Each middleware's `process_request` is called from outermost to innermost
    3. The application handles the request
    4. Each middleware's `process_response` is called from innermost to outermost
    5. The response is sent back to the client

    Examples:
        ```python
        class LoggingMiddleware(Middleware):
            def process_request(self, request):
                print(f"Request: {request.path}")

            def process_response(self, request, response):
                print(f"Response: {response.status_code}")

        app = Plinx()
        app.add_middleware(LoggingMiddleware)
        ```

        Middleware that modifies the request or response:

        ```python
        class AuthMiddleware(Middleware):
            def process_request(self, request):
                request.user = None
                auth_header = request.headers.get("Authorization", "")
                if auth_header.startswith("Bearer "):
                    token = auth_header[7:]
                    request.user = self.get_user_from_token(token)

            def get_user_from_token(self, token):
                # Implementation to validate token and return user
                pass
        ```
    """

    def __init__(
        self,
        app,
    ):
        """
        Initialize the middleware with a WSGI application.

        Args:
            app: A WSGI application (typically a Plinx instance or another middleware)
        """
        self.app = app

    def __call__(
        self,
        environ: dict,
        start_response: callable,
    ):
        """
        WSGI callable interface for the middleware.

        This method makes middleware instances callable according to the WSGI spec,
        allowing them to be used in a WSGI server. It creates a Request object,
        passes it to the application's handle_request method, and returns the
        response.

        Args:
            environ: The WSGI environment dictionary
            start_response: The WSGI start_response callable

        Returns:
            An iterable of bytes representing the response body
        """
        request = Request(environ)

        response = self.app.handle_request(request)

        return response(environ, start_response)

    def add(
        self,
        middleware_cls,
    ):
        """
        Add a new middleware to the stack.

        This method creates an instance of the provided middleware class,
        passing the current middleware instance (or application) as the app parameter.
        This builds up a chain of nested middleware instances.

        Args:
            middleware_cls: A class inheriting from Middleware
        """
        self.app = middleware_cls(self.app)

    def process_request(
        self,
        request: Request,
    ):
        """
        Process the request before it reaches the application.

        Override this method in your middleware subclass to modify or inspect
        the request before it's handled by the application or the next middleware.

        Args:
            request: The WebOb Request object
        """
        pass  # pragma: no cover

    def process_response(
        self,
        request: Request,
        response: Response,
    ):
        """
        Process the response after it's generated by the application.

        Override this method in your middleware subclass to modify or inspect
        the response before it's returned to the client or the previous middleware.

        Args:
            request: The WebOb Request object that generated this response
            response: The Response object to be returned
        """
        pass  # pragma: no cover

    def handle_request(self, request: Request):
        """
        Process a request through this middleware and the wrapped application.

        This method implements the middleware chain by:
        1. Calling this middleware's process_request method
        2. Passing the request to the wrapped application/middleware
        3. Calling this middleware's process_response method with the response

        Args:
            request: The WebOb Request object

        Returns:
            The Response object after processing
        """
        self.process_request(request)
        response = self.app.handle_request(request)
        self.process_response(request, response)

        return response

Functions

__call__(environ, start_response)

WSGI callable interface for the middleware.

This method makes middleware instances callable according to the WSGI spec, allowing them to be used in a WSGI server. It creates a Request object, passes it to the application's handle_request method, and returns the response.

Parameters:

Name	Type	Description	Default
environ	dict	The WSGI environment dictionary	required
start_response	callable	The WSGI start_response callable	required

Returns:

Type	Description
	An iterable of bytes representing the response body

Source code in plinx/middleware.py

def __call__(
    self,
    environ: dict,
    start_response: callable,
):
    """
    WSGI callable interface for the middleware.

    This method makes middleware instances callable according to the WSGI spec,
    allowing them to be used in a WSGI server. It creates a Request object,
    passes it to the application's handle_request method, and returns the
    response.

    Args:
        environ: The WSGI environment dictionary
        start_response: The WSGI start_response callable

    Returns:
        An iterable of bytes representing the response body
    """
    request = Request(environ)

    response = self.app.handle_request(request)

    return response(environ, start_response)

__init__(app)

Initialize the middleware with a WSGI application.

Parameters:

Name	Type	Description	Default
app		A WSGI application (typically a Plinx instance or another middleware)	required

Source code in plinx/middleware.py

def __init__(
    self,
    app,
):
    """
    Initialize the middleware with a WSGI application.

    Args:
        app: A WSGI application (typically a Plinx instance or another middleware)
    """
    self.app = app

add(middleware_cls)

Add a new middleware to the stack.

This method creates an instance of the provided middleware class, passing the current middleware instance (or application) as the app parameter. This builds up a chain of nested middleware instances.

Parameters:

Name	Type	Description	Default
middleware_cls		A class inheriting from Middleware	required

Source code in plinx/middleware.py

def add(
    self,
    middleware_cls,
):
    """
    Add a new middleware to the stack.

    This method creates an instance of the provided middleware class,
    passing the current middleware instance (or application) as the app parameter.
    This builds up a chain of nested middleware instances.

    Args:
        middleware_cls: A class inheriting from Middleware
    """
    self.app = middleware_cls(self.app)

handle_request(request)

Process a request through this middleware and the wrapped application.

This method implements the middleware chain by: 1. Calling this middleware's process_request method 2. Passing the request to the wrapped application/middleware 3. Calling this middleware's process_response method with the response

Parameters:

Name	Type	Description	Default
request	Request	The WebOb Request object	required

Returns:

Type	Description
	The Response object after processing

Source code in plinx/middleware.py

def handle_request(self, request: Request):
    """
    Process a request through this middleware and the wrapped application.

    This method implements the middleware chain by:
    1. Calling this middleware's process_request method
    2. Passing the request to the wrapped application/middleware
    3. Calling this middleware's process_response method with the response

    Args:
        request: The WebOb Request object

    Returns:
        The Response object after processing
    """
    self.process_request(request)
    response = self.app.handle_request(request)
    self.process_response(request, response)

    return response

process_request(request)

Process the request before it reaches the application.

Override this method in your middleware subclass to modify or inspect the request before it's handled by the application or the next middleware.

Parameters:

Name	Type	Description	Default
request	Request	The WebOb Request object	required

Source code in plinx/middleware.py

def process_request(
    self,
    request: Request,
):
    """
    Process the request before it reaches the application.

    Override this method in your middleware subclass to modify or inspect
    the request before it's handled by the application or the next middleware.

    Args:
        request: The WebOb Request object
    """
    pass  # pragma: no cover

process_response(request, response)

Process the response after it's generated by the application.

Override this method in your middleware subclass to modify or inspect the response before it's returned to the client or the previous middleware.

Parameters:

Name	Type	Description	Default
request	Request	The WebOb Request object that generated this response	required
response	Response	The Response object to be returned	required

Source code in plinx/middleware.py

def process_response(
    self,
    request: Request,
    response: Response,
):
    """
    Process the response after it's generated by the application.

    Override this method in your middleware subclass to modify or inspect
    the response before it's returned to the client or the previous middleware.

    Args:
        request: The WebOb Request object that generated this response
        response: The Response object to be returned
    """
    pass  # pragma: no cover

Custom Middleware Examples

Simple Logging Middleware

from plinx.middleware import Middleware

class LoggingMiddleware(Middleware):
    def process_request(self, request):
        print(f"Incoming request: {request.method} {request.path}")

    def process_response(self, request, response):
        print(f"Outgoing response: {response.status_code}")

Request Timer Middleware

import time
from plinx.middleware import Middleware

class TimerMiddleware(Middleware):
    def process_request(self, request):
        request.start_time = time.time()

    def process_response(self, request, response):
        if hasattr(request, "start_time"):
            duration = time.time() - request.start_time
            print(f"Request took {duration:.6f} seconds")
            # Add timing header
            response.headers["X-Request-Duration"] = f"{duration:.6f}"

Authentication Middleware

from plinx.middleware import Middleware

class AuthMiddleware(Middleware):
    def process_request(self, request):
        request.user = None

        # Check for authentication header
        auth_header = request.headers.get("Authorization", "")
        if auth_header.startswith("Bearer "):
            token = auth_header[7:]
            try:
                # In a real app, you'd verify the token
                user_id = self.verify_token(token)
                request.user = {"id": user_id}
            except Exception as e:
                print(f"Auth error: {str(e)}")

    def verify_token(self, token):
        # Simplified demo - in reality, you'd verify with your auth system
        if token == "valid_demo_token":
            return 1
        raise Exception("Invalid token")

CORS Middleware

from plinx.middleware import Middleware

class CORSMiddleware(Middleware):
    def __init__(self, app, allowed_origins=None):
        super().__init__(app)
        self.allowed_origins = allowed_origins or ["*"]

    def process_response(self, request, response):
        origin = request.headers.get("Origin", "")

        # Check if the origin is allowed
        if "*" in self.allowed_origins or origin in self.allowed_origins:
            response.headers["Access-Control-Allow-Origin"] = origin
            response.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, DELETE, OPTIONS"
            response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"

        # Handle preflight OPTIONS requests
        if request.method == "OPTIONS":
            response.status_code = 200
            return response

Using Multiple Middleware

You can add multiple middleware components to a Plinx application:

from plinx import Plinx
from plinx.middleware import Middleware

class Middleware1(Middleware):
    def process_request(self, request):
        print("Middleware 1: process_request")

    def process_response(self, request, response):
        print("Middleware 1: process_response")

class Middleware2(Middleware):
    def process_request(self, request):
        print("Middleware 2: process_request")

    def process_response(self, request, response):
        print("Middleware 2: process_response")

app = Plinx()

# Order matters! Middleware1 will be called before Middleware2 for requests,
# but after Middleware2 for responses
app.add_middleware(Middleware1)
app.add_middleware(Middleware2)

This will produce the following execution order for a request: 1. Middleware1.process_request 2. Middleware2.process_request 3. Handler processes the request 4. Middleware2.process_response 5. Middleware1.process_response

Short-Circuiting Requests

A middleware can short-circuit the request processing by returning a response:

from plinx.middleware import Middleware
from plinx.status_codes import StatusCodes

class AuthRequiredMiddleware(Middleware):
    def process_request(self, request):
        auth_header = request.headers.get("Authorization", "")

        # If no auth header, short-circuit the request
        if not auth_header:
            # Need to access the app to create a response
            response = self.app.handle_request(request)
            response.status_code = StatusCodes.UNAUTHORIZED.value
            response.json = {"error": "Authentication required"}
            return response  # Short-circuit

Error Handling in Middleware

from plinx.middleware import Middleware
from plinx.status_codes import StatusCodes

class ErrorCatchingMiddleware(Middleware):
    def process_request(self, request):
        try:
            # Normal request processing
            pass
        except Exception as e:
            print(f"Request error: {str(e)}")

    def process_response(self, request, response):
        # You can modify the response based on conditions
        if response.status_code >= 500:
            response.json = {
                "error": "An internal error occurred",
                "status_code": response.status_code
            }
            # Log the error
            print(f"Server error occurred: {response.status_code}")