🔍 Code Extractor

ODataBatchV3Request is a specialized OData request handler for constructing and processing OData v3 batch requests using multipart/mixed HTTP format.

Purpose

This class extends ODataRequest to handle batch operations in OData v3 protocol. It serializes multiple queries into a single multipart/mixed HTTP request, sends them as a batch, and deserializes the multipart response back into individual query responses. This is useful for optimizing network traffic by combining multiple OData operations (GET, POST, PUT, DELETE) into a single HTTP request.

Source Code

class ODataBatchV3Request(ODataRequest):
    def build_request(self, query):
        """
        Construct a OData v3 Batch request

        :type query: office365.runtime.queries.batch.BatchQuery
        """
        request = RequestOptions(query.url)
        request.method = HttpMethod.Post
        media_type = "multipart/mixed"
        content_type = "; ".join(
            [media_type, "boundary={0}".format(query.current_boundary)]
        )
        request.ensure_header("Content-Type", content_type)
        request.data = self._prepare_payload(query)
        return request

    def process_response(self, response, query):
        """
        Parses an HTTP response.

        :type response: requests.Response
        :type query: office365.runtime.queries.batch.BatchQuery
        """
        for sub_qry, sub_resp in self._extract_response(response, query):
            sub_resp.raise_for_status()
            super(ODataBatchV3Request, self).process_response(sub_resp, sub_qry)

    def _extract_response(self, response, query):
        """Parses a multipart/mixed response body from the position defined by the context.

        :type response: requests.Response
        :type query: office365.runtime.queries.batch.BatchQuery
        """
        content_type = response.headers["Content-Type"].encode("ascii")
        http_body = b"Content-Type: " + content_type + b"\r\n\r\n" + response.content

        message = message_from_bytes_or_string(http_body)  # type: Message

        query_id = 0
        for raw_response in message.get_payload():
            if raw_response.get_content_type() == "application/http":
                qry = query.ordered_queries[query_id]
                query_id += 1
                yield qry, self._deserialize_response(raw_response)

    def _prepare_payload(self, query):
        """
        Serializes a batch request body.

        :type query: office365.runtime.queries.batch.BatchQuery
        """
        main_message = Message()
        main_message.add_header("Content-Type", "multipart/mixed")
        main_message.set_boundary(query.current_boundary)

        if query.has_change_sets:
            change_set_message = Message()
            change_set_boundary = create_boundary("changeset_", True)
            change_set_message.add_header("Content-Type", "multipart/mixed")
            change_set_message.set_boundary(change_set_boundary)

            for qry in query.change_sets:
                request = qry.build_request()
                message = self._serialize_request(request)
                change_set_message.attach(message)
            main_message.attach(change_set_message)

        for qry in query.get_queries:
            request = qry.build_request()
            message = self._serialize_request(request)
            main_message.attach(message)

        return message_as_bytes_or_string(main_message)

    @staticmethod
    def _normalize_headers(headers_raw):
        """
        :type headers_raw: list[str]
        """
        headers = {}
        for header_line in headers_raw:
            k, v = header_line.split(":", 1)
            headers[k.title()] = v.strip()
        return CaseInsensitiveDict(headers)

    def _deserialize_response(self, raw_response):
        """
        :type raw_response: Message
        """
        response = raw_response.get_payload(decode=True)
        lines = list(filter(None, response.decode("utf-8").split("\r\n")))
        response_status_regex = "^HTTP/1\\.\\d (\\d{3}) (.*)$"
        status_result = re.match(response_status_regex, lines[0])
        status_info = status_result.groups()

        resp = requests.Response()
        resp.status_code = int(status_info[0])
        if status_info[1] == "No Content" or len(lines) < 3:
            resp.headers = self._normalize_headers(lines[1:])
            resp._content = bytes(str("").encode("utf-8"))
        else:
            resp._content = bytes(str(lines[-1]).encode("utf-8"))
            resp.headers = self._normalize_headers(lines[1:-1])
        return resp

    @staticmethod
    def _serialize_request(request):
        """Serializes a part of a batch request to a string. A part can be either a GET request or
            a change set grouping several CUD (create, update, delete) requests.

        :type request: RequestOptions
        """
        eol = "\r\n"
        method = request.method
        if "X-HTTP-Method" in request.headers:
            method = request.headers["X-HTTP-Method"]
        lines = ["{method} {url} HTTP/1.1".format(method=method, url=request.url)] + [
            ":".join(h) for h in request.headers.items()
        ]
        if request.data:
            lines.append(eol)
            lines.append(json.dumps(request.data))
        raw_content = eol + eol.join(lines) + eol
        payload = raw_content.encode("utf-8").lstrip()

        message = Message()
        message.add_header("Content-Type", "application/http")
        message.add_header("Content-Transfer-Encoding", "binary")
        message.set_payload(payload)
        return message

Parameters

Parameter Details

bases: Inherits from ODataRequest, which provides base functionality for OData request handling. The class does not define its own __init__ method, so it uses the parent class constructor.

Return Value

Instantiation returns an ODataBatchV3Request object. The build_request method returns a RequestOptions object configured for batch requests. The process_response method has no return value but processes responses in-place. The _extract_response method yields tuples of (query, response) pairs. The _prepare_payload method returns a serialized batch request body as bytes or string.

Class Interface

Methods

Dependencies

• json
• re
• email.message
• requests
• requests.structures
• office365.runtime.compat
• office365.runtime.http.http_method
• office365.runtime.http.request_options
• office365.runtime.odata.request
• office365.runtime.queries.batch

Required Imports

import json
import re
from email.message import Message
import requests
from requests.structures import CaseInsensitiveDict
from office365.runtime.compat import message_as_bytes_or_string
from office365.runtime.compat import message_from_bytes_or_string
from office365.runtime.http.http_method import HttpMethod
from office365.runtime.http.request_options import RequestOptions
from office365.runtime.odata.request import ODataRequest
from office365.runtime.queries.batch import create_boundary

Usage Example

from office365.runtime.odata.v3.batch_request import ODataBatchV3Request
from office365.runtime.queries.batch import BatchQuery

# Create a batch request handler
batch_request = ODataBatchV3Request()

# Assume we have a BatchQuery object with multiple queries
batch_query = BatchQuery(url='https://api.example.com/$batch')
# Add queries to batch_query (implementation depends on BatchQuery class)

# Build the batch request
request_options = batch_request.build_request(batch_query)

# Send the request (using requests library or similar)
import requests
response = requests.request(
    method=request_options.method,
    url=request_options.url,
    headers=request_options.headers,
    data=request_options.data
)

# Process the batch response
batch_request.process_response(response, batch_query)

Best Practices

• Always ensure the BatchQuery object is properly configured with a valid URL and boundary before calling build_request
• The process_response method will raise HTTP exceptions for failed sub-requests, so wrap calls in try-except blocks
• The class maintains no internal state between method calls, making it safe to reuse for multiple batch operations
• Use this class when you need to combine multiple OData operations to reduce network overhead
• Ensure the OData endpoint supports v3 batch protocol with multipart/mixed format
• The _extract_response method is a generator, yielding query-response pairs for memory efficiency
• Change sets (CUD operations) are automatically grouped together in the batch payload
• GET requests are processed separately from change sets in the batch structure

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class ODataV4BatchRequest 87.9% similar

  A class that handles OData V4 JSON batch requests, allowing multiple OData operations to be bundled and executed in a single HTTP request.

  From: /tf/active/vicechatdev/SPFCsync/venv/lib64/python3.11/site-packages/office365/runtime/odata/v4/batch_request.py

• class ODataRequest 71.8% similar

  ODataRequest is a specialized HTTP request handler for OData protocol operations, extending ClientRequest to build, execute, and process OData-compliant requests with JSON format support.

  From: /tf/active/vicechatdev/SPFCsync/venv/lib64/python3.11/site-packages/office365/runtime/odata/request.py

• class GraphRequest 58.1% similar

  GraphRequest is a specialized OData request class configured for Microsoft Graph API interactions using V4 JSON format.

  From: /tf/active/vicechatdev/SPFCsync/venv/lib64/python3.11/site-packages/office365/graph_request.py

• class BatchCreationResult 57.2% similar

  BatchCreationResult is a data class that inherits from ClientValue, representing the result of a batch creation operation in the Office365 API.

  From: /tf/active/vicechatdev/SPFCsync/venv/lib64/python3.11/site-packages/office365/sharepoint/migrationcenter/common/results/batch_creation.py

• class SharePointRequest 55.2% similar

  SharePointRequest is a specialized OData request class configured to use JSON Light format for SharePoint API interactions.

  From: /tf/active/vicechatdev/SPFCsync/venv/lib64/python3.11/site-packages/office365/sharepoint/request.py