feat(http): implement body inspection handler

packages/http/httpx/kiota_http/kiota_client_factory.py

@@ -11,6 +11,7 @@
 from .middleware import (
     AsyncKiotaTransport,
     BaseMiddleware,
+    BodyInspectionHandler,
     HeadersInspectionHandler,
     MiddlewarePipeline,
     ParametersNameDecodingHandler,
@@ -19,6 +20,7 @@
     UrlReplaceHandler,
 )
 from .middleware.options import (
+    BodyInspectionHandlerOption,
     HeadersInspectionHandlerOption,
     ParametersNameDecodingHandlerOption,
     RedirectHandlerOption,
@@ -91,6 +93,7 @@ def get_default_middleware(options: Optional[dict[str, RequestOption]]) -> list[
         url_replace_handler = UrlReplaceHandler()
         user_agent_handler = UserAgentHandler()
         headers_inspection_handler = HeadersInspectionHandler()
+        body_inspection_handler = BodyInspectionHandler()
 
         if options:
             redirect_handler_options = options.get(RedirectHandlerOption.get_key())
@@ -135,11 +138,18 @@ def get_default_middleware(options: Optional[dict[str, RequestOption]]) -> list[
                     options=headers_inspection_handler_options
                 )
 
-        middleware = [
+            body_inspection_handler_options = options.get(BodyInspectionHandlerOption.get_key())
+            if body_inspection_handler_options and isinstance(
+                body_inspection_handler_options, BodyInspectionHandlerOption
+            ):
+                body_inspection_handler = BodyInspectionHandler(
+                    options=body_inspection_handler_options
+                )
+
+        return [
             redirect_handler, retry_handler, parameters_name_decoding_handler, url_replace_handler,
-            user_agent_handler, headers_inspection_handler
+            user_agent_handler, headers_inspection_handler, body_inspection_handler
         ]
-        return middleware
 
     @staticmethod
     def create_middleware_pipeline(

packages/http/httpx/kiota_http/middleware/__init__.py

@@ -1,4 +1,5 @@
 from .async_kiota_transport import AsyncKiotaTransport
+from .body_inspection_handler import BodyInspectionHandler
 from .headers_inspection_handler import HeadersInspectionHandler
 from .middleware import BaseMiddleware, MiddlewarePipeline
 from .parameters_name_decoding_handler import ParametersNameDecodingHandler

packages/http/httpx/kiota_http/middleware/body_inspection_handler.py

@@ -0,0 +1,90 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation. All Rights Reserved.
+# Licensed under the MIT License.
+# See License in the project root for license information.
+# ------------------------------------
+
+from typing import Optional
+
+import httpx
+
+from .middleware import BaseMiddleware
+from .options import BodyInspectionHandlerOption
+
+BODY_INSPECTION_KEY = "com.microsoft.kiota.handler.body_inspection.enable"
+
+
+class BodyInspectionHandler(BaseMiddleware):
+    """The Body Inspection Handler allows the developer to inspect the body of the
+    request and response.
+    """
+
+    def __init__(
+        self,
+        options: Optional[BodyInspectionHandlerOption] = None,
+    ):
+        """Create an instance of BodyInspectionHandler
+
+        Args:
+            options (BodyInspectionHandlerOption, optional): Default options to apply to the
+            handler. A new BodyInspectionHandlerOption per handler when not provided.
+        """
+        super().__init__()
+        self.options = options if options is not None else BodyInspectionHandlerOption()
+
+    async def send(
+        self, request: httpx.Request, transport: httpx.AsyncBaseTransport
+    ) -> httpx.Response:
+        """To execute the current middleware
+
+        Args:
+            request (httpx.Request): The prepared request object
+            transport (httpx.AsyncBaseTransport): The HTTP transport to use
+
+        Returns:
+            httpx.Response: The response object.
+        """
+        current_options = self._get_current_options(request)
+        span = self._create_observability_span(request, "BodyInspectionHandler_send")
+        span.set_attribute(BODY_INSPECTION_KEY, True)
+        span.end()
+
+        if current_options and current_options.inspect_request_body:
+            content = await request.aread()
+            if content:
+                current_options.request_body = content
+            else:
+                current_options.request_body = None
+
+        response = await super().send(request, transport)
+
+        if current_options and current_options.inspect_response_body:
+            content = await response.aread()
+            if content:
+                current_options.response_body = content
+            else:
+                current_options.response_body = None
+
+        return response
+
+    def _get_current_options(self, request: httpx.Request) -> BodyInspectionHandlerOption:
+        """Returns the options to use for the request. Overrides default options if
+        request options are passed.
+
+        Args:
+            request (httpx.Request): The prepared request object
+
+        Returns:
+            BodyInspectionHandlerOption: The options to be used.
+        """
+        current_options = None
+        request_options = getattr(request, "options", None)
+        if request_options:
+            current_options = request_options.get(BodyInspectionHandlerOption.get_key(), None)
+        if current_options:
+            return current_options
+
+        # Clear body per request
+        self.options.request_body = None
+        self.options.response_body = None
+        return self.options

packages/http/httpx/kiota_http/middleware/options/__init__.py

@@ -1,3 +1,4 @@
+from .body_inspection_handler_option import BodyInspectionHandlerOption
 from .headers_inspection_handler_option import HeadersInspectionHandlerOption
 from .parameters_name_decoding_handler_option import ParametersNameDecodingHandlerOption
 from .redirect_handler_option import RedirectHandlerOption

packages/http/httpx/kiota_http/middleware/options/body_inspection_handler_option.py

@@ -0,0 +1,99 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation. All Rights Reserved.
+# Licensed under the MIT License.
+# See License in the project root for license information.
+# ------------------------------------
+from dataclasses import dataclass
+from io import BytesIO
+from typing import ClassVar, Optional
+
+from kiota_abstractions.request_option import RequestOption
+
+
+@dataclass(eq=False)
+class BodyInspectionHandlerOption(RequestOption):
+    """Config options for the BodyInspectionHandler.
+
+    Args:
+        inspect_request_body (bool, optional): Whether the request body
+        should be inspected. Defaults to False. Note that this setting
+        increases memory usage as the request body is copied in memory.
+        inspect_response_body (bool, optional): Whether the response body
+        should be inspected. Defaults to False. Note that this setting
+        increases memory usage as the response body is copied in memory.
+        request_body (Optional[bytes], optional): The inspected request body bytes.
+        Defaults to None.
+        response_body (Optional[bytes], optional): The inspected response body bytes.
+        Defaults to None.
+    """
+
+    BODY_INSPECTION_HANDLER_OPTION_KEY: ClassVar[str] = "BodyInspectionHandlerOption"
+
+    inspect_request_body: bool = False
+    inspect_response_body: bool = False
+    request_body: Optional[bytes] = None
+    response_body: Optional[bytes] = None
+
+    @staticmethod
+    def get_key() -> str:
+        return BodyInspectionHandlerOption.BODY_INSPECTION_HANDLER_OPTION_KEY
+
+    def get_request_body(self) -> Optional[bytes]:
+        """Gets the request body as bytes.
+
+        Returns:
+            Optional[bytes]: The request body bytes, or None if inspection was
+            disabled or no body was present.
+        """
+        return self.request_body
+
+    def get_response_body(self) -> Optional[bytes]:
+        """Gets the response body as bytes.
+
+        Returns:
+            Optional[bytes]: The response body bytes, or None if inspection was
+            disabled or no body was present.
+        """
+        return self.response_body
+
+    def get_request_body_stream(self) -> Optional[BytesIO]:
+        """Gets the request body as a seekable stream rewound to position 0.
+
+        Callers are responsible for disposing/closing the stream. Note that this stream
+        is a copy of the original request body, which has impact on memory usage.
+
+        Returns:
+            Optional[BytesIO]: A new BytesIO stream of the request body, or None if
+            inspection was disabled or no body was present.
+        """
+        if self.request_body is not None:
+            stream = BytesIO(self.request_body)
+            stream.seek(0)
+            return stream
+        return None
+
+    def get_response_body_stream(self) -> Optional[BytesIO]:
+        """Gets the response body as a seekable stream rewound to position 0.
+
+        Callers are responsible for disposing/closing the stream. Note that this stream
+        is a copy of the original response body, which has impact on memory usage.
+
+        Returns:
+            Optional[BytesIO]: A new BytesIO stream of the response body, or None if
+            inspection was disabled or no body was present.
+        """
+        if self.response_body is not None:
+            stream = BytesIO(self.response_body)
+            stream.seek(0)
+            return stream
+        return None
+
+    @property
+    def request_body_stream(self) -> Optional[BytesIO]:
+        """Stream property for the request body rewound to position 0."""
+        return self.get_request_body_stream()
+
+    @property
+    def response_body_stream(self) -> Optional[BytesIO]:
+        """Stream property for the response body rewound to position 0."""
+        return self.get_response_body_stream()