import logging
from typing import Optional, List, Any, Protocol, runtime_checkable, Type
from http import HTTPStatus
from .signals import dispatcher
from .response import Response
logger = logging.getLogger(__name__)
[docs]
@runtime_checkable
class Middleware(Protocol):
"""
Protocol defining the interface for middleware objects.
Any class implementing these methods can be considered a Middleware.
"""
[docs]
def process_request(self, request: Any) -> Optional['Response']:
"""
Process an incoming request before the view is dispatched.
Can modify the request or return a Response to short-circuit the request processing chain.
Args:
request: The incoming request object.
Returns:
A Response object to stop further processing and return this response immediately,
or None to allow the request to proceed to the next middleware or the view.
"""
pass
[docs]
def process_response(self, request: Any, response: 'Response') -> 'Response':
"""
Process the outgoing response after the view has been executed (or a middleware short-circuited).
Can modify the response. This method is called in reverse order of middleware addition.
Args:
request: The incoming request object (potentially modified by process_request chain).
response: The Response object generated by the view or a preceding middleware.
Returns:
The modified or original Response object. Must always return a Response.
"""
pass
[docs]
class MiddlewareManager:
"""
Manages a list of middleware instances and applies them to requests and responses.
Middlewares are processed in the order they are added for process_request
and in reverse order for process_response.
Integrates SignalDispatcher to emit events during middleware processing.
"""
[docs]
def __init__(self):
"""
Initializes the MiddlewareManager with an empty list of middlewares.
Emits 'middleware_manager_initialized' signal.
"""
self.middlewares: List[Middleware] = []
logger.info("MiddlewareManager initialized.")
dispatcher.send("middleware_manager_initialized", sender=self)
logger.debug("Signal 'middleware_manager_initialized' sent.")
[docs]
def add_middleware(self, middleware: Middleware):
"""
Registers a new middleware component instance.
Middlewares should be instances of classes implementing the Middleware Protocol.
Emits 'middleware_added' signal on success.
Emits 'middleware_add_failed' signal on failure (invalid type).
Args:
middleware: The middleware instance to add.
Raises:
TypeError: If the provided object does not implement the Middleware Protocol
(i.e., missing process_request or process_response methods).
"""
middleware_name = getattr(middleware, '__class__', type(middleware)).__name__
logger.debug(f"Attempting to add middleware: {middleware_name}")
if not isinstance(middleware, Middleware):
logger.error(f"Attempted to add object that does not implement Middleware Protocol: {middleware_name}")
dispatcher.send("middleware_add_failed", sender=self, middleware_instance=middleware, middleware_name=middleware_name, error_type="invalid_type")
logger.debug(f"Signal 'middleware_add_failed' (invalid_type) sent for '{middleware_name}'.")
raise TypeError("Middleware object must implement the Middleware Protocol (have callable 'process_request' and 'process_response' methods).")
self.middlewares.append(middleware)
logger.info(f"Middleware added successfully: {middleware_name}. Total middlewares: {len(self.middlewares)}")
dispatcher.send("middleware_added", sender=self, middleware_instance=middleware, middleware_name=middleware_name)
logger.debug(f"Signal 'middleware_added' sent for '{middleware_name}'.")
[docs]
def remove_middleware(self, middleware: Middleware):
"""
Removes a specific middleware instance from the manager.
Emits 'middleware_removed' signal on success.
Emits 'middleware_remove_failed' signal on failure (not found).
Args:
middleware: The middleware instance to remove.
"""
middleware_name = getattr(middleware, '__class__', type(middleware)).__name__
logger.debug(f"Attempting to remove middleware instance: {middleware_name}")
try:
self.middlewares.remove(middleware)
logger.info(f"Middleware instance removed: {middleware_name}. Total middlewares: {len(self.middlewares)}")
dispatcher.send("middleware_removed", sender=self, middleware_instance=middleware, middleware_name=middleware_name)
logger.debug(f"Signal 'middleware_removed' sent for '{middleware_name}'.")
except ValueError:
logger.warning(f"Middleware instance not found for removal: {middleware_name}")
dispatcher.send("middleware_remove_failed", sender=self, middleware_instance=middleware, middleware_name=middleware_name, error_type="not_found")
logger.debug(f"Signal 'middleware_remove_failed' (not_found) sent for '{middleware_name}'.")
[docs]
def remove_middleware_by_class(self, middleware_class: Type[Middleware]):
"""
Removes all instances of a given middleware class from the chain.
Emits 'middleware_removed_by_class' signal on success (if any were removed).
Emits 'middleware_remove_by_class_failed' signal on failure (invalid class type).
Args:
middleware_class: The class of the middleware instances to remove.
Should be a class that implements the Middleware Protocol.
"""
middleware_class_name = getattr(middleware_class, '__name__', 'N/A')
if not isinstance(middleware_class, type) or not issubclass(middleware_class, Middleware):
logger.error(f"Attempted to remove middleware by non-Middleware class type: {middleware_class_name}")
dispatcher.send("middleware_remove_by_class_failed", sender=self, middleware_class=middleware_class, middleware_class_name=middleware_class_name, error_type="invalid_type")
logger.debug(f"Signal 'middleware_remove_by_class_failed' (invalid_type) sent for class '{middleware_class_name}'.")
raise TypeError("Provided middleware_class must be a class that implements the Middleware Protocol.")
logger.debug(f"Attempting to remove all instances of middleware class: {middleware_class_name}")
initial_count = len(self.middlewares)
removed_instances = [m for m in self.middlewares if isinstance(m, middleware_class)]
self.middlewares = [m for m in self.middlewares if not isinstance(m, middleware_class)]
removed_count = initial_count - len(self.middlewares)
if removed_count > 0:
logger.info(f"Removed {removed_count} instance(s) of middleware class: {middleware_class_name}. Total middlewares: {len(self.middlewares)}")
dispatcher.send("middleware_removed_by_class", sender=self, middleware_class=middleware_class, middleware_class_name=middleware_class_name, removed_count=removed_count, removed_instances=removed_instances)
logger.debug(f"Signal 'middleware_removed_by_class' sent for class '{middleware_class_name}'.")
else:
logger.warning(f"No instance of middleware class {middleware_class_name} found for removal.")
[docs]
def process_request(self, request: Any) -> Optional[Response]:
"""
Applies the process_request method of each middleware in the order they were added.
Stops the chain and returns a Response object if any middleware returns one.
Exceptions raised by middlewares are caught and result in a 500 Internal Server Error response.
Emits 'middleware_request_processing_started', 'middleware_request_processing_finished',
'middleware_process_request_started', 'middleware_process_request_succeeded',
'middleware_process_request_failed', and 'middleware_request_short_circuited' signals.
Args:
request: The incoming request object. This object is passed through the chain
and can be modified by middlewares (e.g., adding context, user).
Returns:
A Response object if a middleware short-circuits the chain or an error occurs,
otherwise returns None to indicate that the request should proceed to the view.
"""
request_method = getattr(request, 'method', 'N/A')
request_path = getattr(request, 'path', 'N/A')
logger.debug(f"MiddlewareManager: Starting request middleware chain for {request_method} {request_path}.")
dispatcher.send("middleware_request_processing_started", sender=self, request=request, method=request_method, path=request_path)
logger.debug(f"Signal 'middleware_request_processing_started' sent for {request_method} {request_path}.")
final_response = None
for middleware in self.middlewares:
middleware_name = getattr(middleware, '__class__', type(middleware)).__name__
logger.debug(f"MiddlewareManager: Processing request with {middleware_name}")
dispatcher.send("middleware_process_request_started", sender=self, middleware_instance=middleware, middleware_name=middleware_name, request=request)
logger.debug(f"Signal 'middleware_process_request_started' sent for {middleware_name}.")
try:
response = middleware.process_request(request)
if response is not None:
logger.info(f"Request short-circuited by middleware: {middleware_name} (Status: {getattr(response, 'status_code', 'N/A')})")
if not isinstance(response, Response):
logger.error(f"Middleware {middleware_name} returned unexpected type {type(response)} in process_request for {request_method} {request_path}. Expected Response. Returning 500.")
dispatcher.send("middleware_process_request_failed", sender=self, middleware_instance=middleware, middleware_name=middleware_name, request=request, error_type="invalid_response_type", returned_type=type(response))
logger.debug(f"Signal 'middleware_process_request_failed' (invalid_response_type) sent for {middleware_name}.")
final_response = Response(body=b"Internal Server Error: Invalid middleware response type.", status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, headers={'Content-Type': 'text/plain'})
else:
final_response = response
dispatcher.send("middleware_request_short_circuited", sender=self, middleware_instance=middleware, middleware_name=middleware_name, request=request, response=final_response)
logger.debug(f"Signal 'middleware_request_short_circuited' sent by {middleware_name}.")
break
else:
logger.debug(f"Middleware {middleware_name} processed request, continuing chain.")
dispatcher.send("middleware_process_request_succeeded", sender=self, middleware_instance=middleware, middleware_name=middleware_name, request=request)
logger.debug(f"Signal 'middleware_process_request_succeeded' sent for {middleware_name}.")
except Exception as e:
logger.exception(f"Error in process_request of {middleware_name} for {request_method} {request_path}: {e}")
dispatcher.send("middleware_process_request_failed", sender=self, middleware_instance=middleware, middleware_name=middleware_name, request=request, error_type="exception", exception=e)
logger.debug(f"Signal 'middleware_process_request_failed' (exception) sent for {middleware_name}.")
final_response = Response(body=b"Internal Server Error: Middleware processing failed.", status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, headers={'Content-Type': 'text/plain'})
break
if final_response is None:
logger.debug("MiddlewareManager: Request passed through all request middlewares without short-circuit. Proceeding to view.")
dispatcher.send("middleware_request_processing_finished", sender=self, request=request, method=request_method, path=request_path)
logger.debug(f"Signal 'middleware_request_processing_finished' sent for {request_method} {request_path}.")
return final_response
[docs]
def process_response(self, request: Any, response: Response) -> Response:
"""
Applies the process_response method of each middleware in reverse order of addition.
Starts with the response generated by the view or a short-circuiting middleware.
Each middleware receives the output of the previous one.
Exceptions raised by middlewares are caught and logged, but the response chain attempts to continue.
Emits 'middleware_response_processing_started', 'middleware_response_processing_finished',
'middleware_process_response_started', 'middleware_process_response_succeeded',
and 'middleware_process_response_failed' signals.
Args:
request: The incoming request object (potentially modified by process_request chain).
response: The initial Response object generated by the view or a short-circuiting middleware.
Returns:
The final Response object after all response middlewares have been applied.
"""
request_method = getattr(request, 'method', 'N/A')
request_path = getattr(request, 'path', 'N/A')
logger.debug(f"MiddlewareManager: Starting response middleware chain for {request_method} {request_path}.")
dispatcher.send("middleware_response_processing_started", sender=self, request=request, method=request_method, path=request_path, initial_response=response)
logger.debug(f"Signal 'middleware_response_processing_started' sent for {request_method} {request_path}.")
processed_response = response
for middleware in reversed(self.middlewares):
middleware_name = getattr(middleware, '__class__', type(middleware)).__name__
logger.debug(f"MiddlewareManager: Processing response with {middleware_name}")
dispatcher.send("middleware_process_response_started", sender=self, middleware_instance=middleware, middleware_name=middleware_name, request=request, current_response=processed_response)
logger.debug(f"Signal 'middleware_process_response_started' sent for {middleware_name}.")
try:
middleware_response = middleware.process_response(request, processed_response)
if not isinstance(middleware_response, Response):
logger.error(f"Response Middleware {middleware_name} returned unexpected type {type(middleware_response)} in process_response for {request_method} {request_path}. Expected Response. Attempting to continue with previous response.")
dispatcher.send("middleware_process_response_failed", sender=self, middleware_instance=middleware, middleware_name=middleware_name, request=request, current_response=processed_response, error_type="invalid_response_type", returned_type=type(middleware_response))
logger.debug(f"Signal 'middleware_process_response_failed' (invalid_response_type) sent for {middleware_name}.")
else:
processed_response = middleware_response
logger.debug(f"Response processed by: {middleware_name}")
dispatcher.send("middleware_process_response_succeeded", sender=self, middleware_instance=middleware, middleware_name=middleware_name, request=request, processed_response=processed_response)
logger.debug(f"Signal 'middleware_process_response_succeeded' sent for {middleware_name}.")
except Exception as e:
logger.exception(f"Error in process_response of {middleware_name} for {request_method} {request_path}: {e}")
dispatcher.send("middleware_process_response_failed", sender=self, middleware_instance=middleware, middleware_name=middleware_name, request=request, current_response=processed_response, error_type="exception", exception=e)
logger.debug(f"Signal 'middleware_process_response_failed' (exception) sent for {middleware_name}.")
logger.debug(f"MiddlewareManager: Response passed through all response middlewares. Finalizing response for {request_method} {request_path}.")
dispatcher.send("middleware_response_processing_finished", sender=self, request=request, method=request_method, path=request_path, final_response=processed_response)
logger.debug(f"Signal 'middleware_response_processing_finished' sent for {request_method} {request_path}.")
return processed_response