🔍 Code Extractor

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.

Purpose

This class manages the complete lifecycle of OData requests including building HTTP requests from queries, processing responses, and mapping JSON data to client objects. It handles various query types (CRUD operations, service operations, functions) and supports different OData JSON formats (JsonLight, etc.). The class ensures proper content-type headers, normalizes payloads, and transforms OData JSON responses into strongly-typed client objects.

Source Code

class ODataRequest(ClientRequest):
    def __init__(self, json_format):
        # type: (ODataJsonFormat) -> None
        """Creates OData request"""
        super(ODataRequest, self).__init__()
        self._default_json_format = json_format
        self.beforeExecute += self._ensure_json_format

    @property
    def json_format(self):
        return self._default_json_format

    def build_request(self, query):
        # type: (ClientQuery) -> RequestOptions
        """Builds a request"""
        request = RequestOptions(query.url)
        request.method = HttpMethod.Get
        if isinstance(query, DeleteEntityQuery):
            request.method = HttpMethod.Post
        elif isinstance(
            query, (CreateEntityQuery, UpdateEntityQuery, ServiceOperationQuery)
        ):
            request.method = HttpMethod.Post
            if query.parameters_type is not None:
                request.data = self._build_payload(query)
        return request

    def process_response(self, response, query):
        # type: (requests.Response, ClientQuery) -> None
        json_format = copy.deepcopy(self.json_format)
        return_type = query.return_type
        if return_type is None:
            return

        if isinstance(return_type, ClientObject):
            return_type.clear()

        if (
            response.headers.get("Content-Type", "").lower().split(";")[0]
            != "application/json"
        ):
            if isinstance(return_type, ClientResult):
                return_type.set_property("__value", response.content)
        else:
            if isinstance(json_format, JsonLightFormat):
                if isinstance(query, (ServiceOperationQuery, FunctionQuery)):
                    json_format.function = query.name

            self.map_json(response.json(), return_type, json_format)

    def map_json(self, json, return_type, json_format=None):
        """
        :type json: any
        :type return_type: ClientValue or ClientResult or ClientObject
        :type json_format: office365.runtime.odata.json_format.ODataJsonFormat
        """
        if json_format is None:
            json_format = self.json_format

        if json and return_type is not None:
            for k, v in self._next_property(json, json_format):
                return_type.set_property(k, v, False)

    def _next_property(self, json, json_format):
        """
        :type json: Any
        :type json_format: office365.runtime.odata.json_format.ODataJsonFormat
        """
        if isinstance(json_format, JsonLightFormat):
            json = json.get(json_format.security, json)
            json = json.get(json_format.function, json)

        if isinstance(json, dict):
            next_link_url = json.get(json_format.collection_next, None)
            json = json.get(json_format.collection, json)
            if next_link_url:
                yield "__nextLinkUrl", next_link_url

            if isinstance(json, list):
                for index, item in enumerate(json):
                    if isinstance(item, dict):
                        item = {k: v for k, v in self._next_property(item, json_format)}
                    yield index, item
            elif isinstance(json, dict):
                for name, value in json.items():
                    if isinstance(json_format, JsonLightFormat):
                        is_valid = name != "__metadata" and not (
                            isinstance(value, dict) and "__deferred" in value
                        )
                    else:
                        is_valid = "@odata" not in name

                    if is_valid:
                        if isinstance(value, dict):
                            value = {
                                k: v for k, v in self._next_property(value, json_format)
                            }
                        yield name, value
            else:
                yield "__value", json
        elif json is not None:
            yield "__value", json

    def _build_payload(self, query):
        # type: (ClientQuery) -> dict|list
        """Normalizes OData request payload"""

        def _normalize_payload(payload):
            # type: (ClientObject|ClientValue|dict|list) -> dict|list
            if isinstance(payload, (ClientObject, ClientValue)):
                return payload.to_json(self._default_json_format)
            elif isinstance(payload, dict):
                return {
                    k: _normalize_payload(v)
                    for k, v in payload.items()
                    if v is not None
                }
            elif isinstance(payload, list):
                return [_normalize_payload(item) for item in payload]
            return payload

        json = _normalize_payload(query.parameters_type)
        if (
            isinstance(query, ServiceOperationQuery)
            and query.parameters_name is not None
        ):
            json = {query.parameters_name: json}
        return json

    def _ensure_json_format(self, request):
        # type: (RequestOptions) -> None
        media_type = self.json_format.media_type
        request.ensure_header("Content-Type", media_type)
        request.ensure_header("Accept", media_type)

Parameters

Name	Type	Default	Kind
bases	ClientRequest	-

Parameter Details

json_format: An ODataJsonFormat instance that defines the JSON serialization/deserialization format for OData requests and responses. This determines how data is structured (e.g., JsonLightFormat for OData v3). The format is used throughout the request lifecycle for payload building and response parsing.

Return Value

The constructor returns an ODataRequest instance configured with the specified JSON format. The build_request method returns a RequestOptions object containing the HTTP method, URL, and payload. The process_response method returns None but modifies the query's return_type object in-place. The map_json method returns None but populates the return_type object with parsed data.

Class Interface

Methods

Attributes

Name	Type	Description	Scope
_default_json_format	ODataJsonFormat	Stores the JSON format configuration used for serialization and deserialization throughout the request lifecycle	instance

Dependencies

• requests
• copy

Required Imports

import copy
import requests
from office365.runtime.client_object import ClientObject
from office365.runtime.client_request import ClientRequest
from office365.runtime.client_result import ClientResult
from office365.runtime.client_value import ClientValue
from office365.runtime.http.http_method import HttpMethod
from office365.runtime.http.request_options import RequestOptions
from office365.runtime.odata.json_format import ODataJsonFormat
from office365.runtime.odata.v3.json_light_format import JsonLightFormat
from office365.runtime.queries.client_query import ClientQuery
from office365.runtime.queries.create_entity import CreateEntityQuery
from office365.runtime.queries.delete_entity import DeleteEntityQuery
from office365.runtime.queries.function import FunctionQuery
from office365.runtime.queries.service_operation import ServiceOperationQuery
from office365.runtime.queries.update_entity import UpdateEntityQuery

Usage Example

from office365.runtime.odata.request import ODataRequest
from office365.runtime.odata.v3.json_light_format import JsonLightFormat
from office365.runtime.queries.service_operation import ServiceOperationQuery
from office365.runtime.client_result import ClientResult

# Create OData request handler with JSON Light format
json_format = JsonLightFormat()
odata_request = ODataRequest(json_format)

# Build a request from a query
query = ServiceOperationQuery(url='https://api.example.com/service', name='GetData')
request_options = odata_request.build_request(query)

# Process a response (typically called internally)
import requests
response = requests.get(request_options.url)
result = ClientResult()
query.return_type = result
odata_request.process_response(response, query)

# Map JSON directly to an object
data = {'name': 'John', 'age': 30}
result_obj = ClientResult()
odata_request.map_json(data, result_obj)

Best Practices

• Always instantiate with an appropriate ODataJsonFormat implementation matching your OData version
• The class automatically registers _ensure_json_format as a beforeExecute handler during initialization
• Use build_request to create RequestOptions before executing HTTP calls
• Call process_response after receiving HTTP responses to automatically populate return_type objects
• The class modifies return_type objects in-place, so ensure query.return_type is set before calling process_response
• For custom JSON mapping, use map_json with an optional json_format parameter to override the default
• The class handles pagination through __nextLinkUrl extraction from OData responses
• Payload normalization automatically converts ClientObject and ClientValue instances to JSON
• The class filters out OData metadata fields (__metadata, __deferred, @odata.*) during response processing
• State is managed through the _default_json_format attribute which should not be modified after initialization

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class ODataV4BatchRequest 78.0% 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 GraphRequest 73.9% 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 SharePointRequest 72.6% 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

• class ODataBatchV3Request 71.8% similar

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

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

• class SearchRequest_v1 60.5% similar

  A class representing a search request formatted as a JSON blob for Microsoft Graph API search operations, encapsulating query parameters and search configuration.

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

class ODataRequest(ClientRequest):
    def __init__(self, json_format):
        # type: (ODataJsonFormat) -> None
        """Creates OData request"""
        super(ODataRequest, self).__init__()
        self._default_json_format = json_format
        self.beforeExecute += self._ensure_json_format

    @property
    def json_format(self):
        return self._default_json_format

    def build_request(self, query):
        # type: (ClientQuery) -> RequestOptions
        """Builds a request"""
        request = RequestOptions(query.url)
        request.method = HttpMethod.Get
        if isinstance(query, DeleteEntityQuery):
            request.method = HttpMethod.Post
        elif isinstance(
            query, (CreateEntityQuery, UpdateEntityQuery, ServiceOperationQuery)
        ):
            request.method = HttpMethod.Post
            if query.parameters_type is not None:
                request.data = self._build_payload(query)
        return request

    def process_response(self, response, query):
        # type: (requests.Response, ClientQuery) -> None
        json_format = copy.deepcopy(self.json_format)
        return_type = query.return_type
        if return_type is None:
            return

        if isinstance(return_type, ClientObject):
            return_type.clear()

        if (
            response.headers.get("Content-Type", "").lower().split(";")[0]
            != "application/json"
        ):
            if isinstance(return_type, ClientResult):
                return_type.set_property("__value", response.content)
        else:
            if isinstance(json_format, JsonLightFormat):
                if isinstance(query, (ServiceOperationQuery, FunctionQuery)):
                    json_format.function = query.name

            self.map_json(response.json(), return_type, json_format)

    def map_json(self, json, return_type, json_format=None):
        """
        :type json: any
        :type return_type: ClientValue or ClientResult or ClientObject
        :type json_format: office365.runtime.odata.json_format.ODataJsonFormat
        """
        if json_format is None:
            json_format = self.json_format

        if json and return_type is not None:
            for k, v in self._next_property(json, json_format):
                return_type.set_property(k, v, False)

    def _next_property(self, json, json_format):
        """
        :type json: Any
        :type json_format: office365.runtime.odata.json_format.ODataJsonFormat
        """
        if isinstance(json_format, JsonLightFormat):
            json = json.get(json_format.security, json)
            json = json.get(json_format.function, json)

        if isinstance(json, dict):
            next_link_url = json.get(json_format.collection_next, None)
            json = json.get(json_format.collection, json)
            if next_link_url:
                yield "__nextLinkUrl", next_link_url

            if isinstance(json, list):
                for index, item in enumerate(json):
                    if isinstance(item, dict):
                        item = {k: v for k, v in self._next_property(item, json_format)}
                    yield index, item
            elif isinstance(json, dict):
                for name, value in json.items():
                    if isinstance(json_format, JsonLightFormat):
                        is_valid = name != "__metadata" and not (
                            isinstance(value, dict) and "__deferred" in value
                        )
                    else:
                        is_valid = "@odata" not in name

                    if is_valid:
                        if isinstance(value, dict):
                            value = {
                                k: v for k, v in self._next_property(value, json_format)
                            }
                        yield name, value
            else:
                yield "__value", json
        elif json is not None:
            yield "__value", json

    def _build_payload(self, query):
        # type: (ClientQuery) -> dict|list
        """Normalizes OData request payload"""

        def _normalize_payload(payload):
            # type: (ClientObject|ClientValue|dict|list) -> dict|list
            if isinstance(payload, (ClientObject, ClientValue)):
                return payload.to_json(self._default_json_format)
            elif isinstance(payload, dict):
                return {
                    k: _normalize_payload(v)
                    for k, v in payload.items()
                    if v is not None
                }
            elif isinstance(payload, list):
                return [_normalize_payload(item) for item in payload]
            return payload

        json = _normalize_payload(query.parameters_type)
        if (
            isinstance(query, ServiceOperationQuery)
            and query.parameters_name is not None
        ):
            json = {query.parameters_name: json}
        return json

    def _ensure_json_format(self, request):
        # type: (RequestOptions) -> None
        media_type = self.json_format.media_type
        request.ensure_header("Content-Type", media_type)
        request.ensure_header("Accept", media_type)