Batch Requests Samples

samples/batch_requests/batch_get_response_body_as_stream.py

@@ -0,0 +1,26 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+#pylint: disable=undefined-variable
+""" Demonstrate getting response body as stream in Batch Responses"""
+batch_request_content = {
+    batch_request_item1.id: batch_request_item1,
+    batch_request_item2.id: batch_request_item2
+}
+
+batch_content = BatchRequestContent(batch_request_content)
+
+
+async def main():
+    batch_response_content = await batch_request_builder.post(batch_request_content=batch_content)
+
+    try:
+        stream_response = batch_response_content.get_response_stream_by_id(batch_request_item1.id)
+        print(f"Stream Response: {stream_response}")
+        print(f"Stream Response Content: {stream_response.read()}")
+    except AttributeError as e:
+        print(f"Error getting response by ID: {e}")
+
+
+asyncio.run(main())

samples/batch_requests/batch_request_with_collection.py

@@ -0,0 +1,99 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+#pylint: disable=undefined-variable
+"""Demonstrates using the Batch request with Collection"""
+import asyncio
+
+from urllib.request import Request
+from kiota_abstractions.request_information import RequestInformation
+
+from msgraph import GraphServiceClient
+
+from kiota_abstractions.headers_collection import HeadersCollection as RequestHeaders
+from msgraph_core.requests.batch_request_item import BatchRequestItem
+
+from msgraph_core.requests.batch_request_content import BatchRequestContent
+from msgraph_core.requests.batch_request_content_collection import BatchRequestContentCollection
+from msgraph_core.requests.batch_request_builder import BatchRequestBuilder
+# Create a client
+# code to create graph client
+
+user_client = GraphServiceClient(credentials=token, scopes=graph_scopes)
+
+# Create a request adapter from the client
+request_adapter = user_client.request_adapter
+
+# Create an instance of BatchRequestBuilder
+batch_request_builder = BatchRequestBuilder(request_adapter)
+
+# Create some BatchRequestItems
+
+request_info1 = RequestInformation()
+request_info1.http_method = "GET"
+request_info1.url = "/me"
+request_info1.headers = RequestHeaders()
+request_info1.headers.add("Content-Type", "application/json")
+
+request_info2 = RequestInformation()
+request_info2.http_method = "GET"
+request_info2.url = "/users"
+request_info2.headers = RequestHeaders()
+request_info2.headers.add("Content-Type", "application/json")
+
+request_info3 = RequestInformation()
+request_info3.http_method = "GET"
+request_info3.url = "/me"
+request_info3.headers = RequestHeaders()
+request_info3.headers.add("Content-Type", "application/json")
+
+# Create BatchRequestItem instances
+batch_request_item1 = BatchRequestItem(request_information=request_info1)
+batch_request_item2 = BatchRequestItem(request_information=request_info2)
+
+# Add a request using RequestInformation directly
+batch_request_content.add_request_information(request_info1)
+print(
+    f"Number of requests after adding request using RequestInformation: {len(batch_request_content.requests)}"
+)
+print("------------------------------------------------------------------------------------")
+# Create an instance of BatchRequestContentCollection
+collection = BatchRequestContentCollection()
+# Add request items to the collection
+batch_request_item_to_add = BatchRequestItem(request_information=request_info3)
+batch_request_item_to_add1 = BatchRequestItem(request_information=request_info3)
+batch_request_item_to_add2 = BatchRequestItem(request_information=request_info3)
+batch_request_item_to_add3 = BatchRequestItem(request_information=request_info3)
+batch_request_item_to_add4 = BatchRequestItem(request_information=request_info3)
+
+collection.add_batch_request_item(batch_request_item_to_add)
+collection.add_batch_request_item(batch_request_item_to_add1)
+collection.add_batch_request_item(batch_request_item_to_add2)
+collection.add_batch_request_item(batch_request_item_to_add3)
+collection.add_batch_request_item(batch_request_item_to_add4)
+
+# Print the current batch requests
+print("Current Batch Requests:")
+for request in collection.current_batch.requests:
+    print(f"Request ID: {request.id}, Status Code: {request.headers}")
+
+# Remove a request item from the collection
+collection.remove_batch_request_item(batch_request_item_to_add.id)
+print(f"Items left in the batch after removal: {len(collection.current_batch.requests)}")
+
+
+# post a collection
+async def main():
+
+    batch_response_content = await batch_request_builder.post(batch_request_content=collection)
+    responses = batch_response_content.get_responses()
+    for item in responses:
+        for item_body in item.responses:
+            print(f"Item: {item_body.id}, Status Code: {item_body.status}")
+            print(f"body: {item_body.body} Item headers: {item_body.headers} ")
+            print("-----------------------------------------------")
+
+
+# Run the main function
+asyncio.run(main())

samples/batch_requests/batch_request_with_content.py

@@ -0,0 +1,68 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+#pylint: disable=undefined-variable
+"""Demonstrates using the Batch request with content"""
+import asyncio
+
+from kiota_abstractions.request_information import RequestInformation
+from kiota_abstractions.method import Method
+from kiota_abstractions.headers_collection import HeadersCollection as RequestHeaders
+
+from msgraph_core.requests.batch_request_item import BatchRequestItem
+from msgraph_core.requests.batch_request_content import BatchRequestContent
+from msgraph_core.requests.batch_request_content_collection import BatchRequestContentCollection
+
+from msgraph_core.requests.batch_response_content import BatchResponseContent
+from msgraph_core.requests.batch_response_content_collection import BatchResponseContentCollection
+from msgraph_core.requests.batch_request_builder import BatchRequestBuilder
+
+# Create a client
+# code to create a graph client
+user_client = GraphServiceClient(credentials=token, scopes=graph_scopes)
+
+# Create a  request adapter from the clinet
+request_adapter = user_client.request_adapter
+
+# Create an instance of BatchRequestBuilder
+batch_request_builder = BatchRequestBuilder(request_adapter)
+
+# Create batch Items
+request_info1 = RequestInformation()
+request_info1.http_method = "GET"
+request_info1.url = "https://graph.microsoft.com/v1.0/me"
+request_info1.url = "/me"
+
+request_info1.headers = RequestHeaders()
+request_info1.headers.add("Content-Type", "application/json")
+
+request_info2 = RequestInformation()
+request_info2.http_method = "GET"
+request_info2.url = "/users"
+request_info2.headers = RequestHeaders()
+request_info2.headers.add("Content-Type", "application/json")
+
+batch_request_item1 = BatchRequestItem(request_information=request_info1)
+batch_request_item2 = BatchRequestItem(request_information=request_info2)
+
+# Create a batch request content
+batch_request_content = {
+    batch_request_item1.id: batch_request_item1,
+    batch_request_item2.id: batch_request_item2
+}
+batch_content = BatchRequestContent(batch_request_content)
+
+
+async def main():
+    batch_response_content = await batch_request_builder.post(batch_request_content=batch_content)
+
+    # Print the batch response content
+    print(f"Batch Response Content: {batch_response_content.responses}")
+    for response in batch_response_content.responses:
+        print(f"Request ID: {response.id}, Status: {response.status}")
+        print(f"Response body: {response.body}, headers: {response.headers}")
+        print("-------------------------------------------------------------")
+
+
+asyncio.run(main())

samples/batch_requests/batch_request_with_custom_error_class.py

@@ -0,0 +1,101 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+#pylint: disable=undefined-variable
+"""Demonstrates using the Batch request with Custom Error Class"""
+
+import asyncio
+
+from kiota_abstractions.request_adapter import RequestAdapter
+from kiota_abstractions.serialization import Parsable
+from kiota_abstractions.request_information import RequestInformation
+from kiota_abstractions.method import Method
+from kiota_abstractions.headers_collection import HeadersCollection as RequestHeaders
+
+from msgraph import GraphServiceClient
+
+from msgraph_core.requests.batch_request_item import BatchRequestItem
+from msgraph_core.requests.batch_request_content import BatchRequestContent
+
+from msgraph_core.requests.batch_response_content import BatchResponseContent
+from msgraph_core.requests.batch_request_builder import BatchRequestBuilder
+# create client
+# code to create client
+user_client = GraphServiceClient(credentials=token, scopes=graph_scopes)
+
+
+# Create an Error map Parsable or import it from wherever you have it
+class CustomError(Parsable):
+
+    def __init__(self) -> None:
+        self.error_code: str = None
+        self.message: str = None
+
+    @staticmethod
+    def not_found() -> 'CustomError':
+        error = CustomError()
+        error.error_code = "404"
+        error.message = "Resource not found"
+        return error
+
+
+# Create a request adapter from client
+request_adapter = user_client.request_adapter
+
+# Create an instance of BatchRequestBuilder
+batch_request_builder = BatchRequestBuilder(request_adapter)
+
+# Create batch Items
+request_info1 = RequestInformation()
+request_info1.http_method = "GET"
+request_info1.url = "https://graph.microsoft.com/v1.0/me"
+request_info1.url = "/me"
+
+request_info1.headers = RequestHeaders()
+request_info1.headers.add("Content-Type", "application/json")
+
+request_info2 = RequestInformation()
+request_info2.http_method = "GET"
+request_info2.url = "/users"
+request_info2.headers = RequestHeaders()
+request_info2.headers.add("Content-Type", "application/json")
+
+# get user who does not exist to test 404 in error map
+request_info3 = RequestInformation()
+request_info3.http_method = "GET"
+request_info3.url = "/users/random-id"
+request_info3.headers = RequestHeaders()
+request_info3.headers.add("Content-Type", "application/json")
+
+# bacth request items to be added to content
+batch_request_item1 = BatchRequestItem(request_information=request_info1)
+batch_request_item2 = BatchRequestItem(request_information=request_info2)
+batch_request_item3 = BatchRequestItem(request_information=request_info3)
+
+# Create a BatchRequestContent
+batch_request_content = {
+    batch_request_item1.id: batch_request_item1,
+    batch_request_item2.id: batch_request_item2
+}
+
+batch_content = BatchRequestContent(batch_request_content)
+
+
+# Function to demonstrate the usage of BatchRequestBuilder
+async def main():
+    error_map = {"400": CustomError, "404": CustomError.not_found}
+
+    batch_response_content = await batch_request_builder.post(
+        batch_request_content=batch_content, error_map=error_map
+    )
+
+    # Print the batch response content
+    print(f"Batch Response Content: {batch_response_content.responses}")
+    for response in batch_response_content.responses:
+        print(f"Request ID: {response.id}, Status: {response.status}")
+        print(f"Response body: {response.body}, headers: {response.headers}")
+        print("-------------------------------------------------------------")
+
+
+asyncio.run(main())

samples/batch_requests/batch_request_with_parsable_as_response_type.py

@@ -0,0 +1,74 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+#pylint: disable=undefined-variable
+"""Demonstrates using the Batch request with Parsable Resposnse Type"""
+import asyncio
+
+from kiota_abstractions.request_adapter import RequestAdapter
+from kiota_abstractions.request_information import RequestInformation
+from kiota_abstractions.method import Method
+from kiota_abstractions.headers_collection import HeadersCollection as RequestHeaders
+
+from msgraph_core.requests.batch_request_item import BatchRequestItem
+from msgraph_core.requests.batch_request_content import BatchRequestContent
+
+from msgraph_core.requests.batch_request_builder import BatchRequestBuilder
+
+# import User model to serialize to
+from msgraph.generated.models.user import User
+# Create a client
+# code to create graph client
+user_client = GraphServiceClient(credentials=token, scopes=graph_scopes)
+
+print(f"Graph Scopes: {graph_scopes}")
+
+# Create a  request adapter from the client
+request_adapter = user_client.request_adapter
+
+# Create an instance of BatchRequestBuilder
+batch_request_builder = BatchRequestBuilder(request_adapter)
+
+# Create batch Items
+request_info1 = RequestInformation()
+request_info1.http_method = "GET"
+request_info1.url = "/users/<user-id-1>"
+
+request_info1.headers = RequestHeaders()
+request_info1.headers.add("Content-Type", "application/json")
+
+request_info2 = RequestInformation()
+request_info2.http_method = "GET"
+request_info2.url = "/users/<user-id-2>"
+request_info2.headers = RequestHeaders()
+request_info2.headers.add("Content-Type", "application/json")
+
+# bacth request items to be added to content
+batch_request_item1 = BatchRequestItem(request_information=request_info1)
+batch_request_item2 = BatchRequestItem(request_information=request_info2)
+
+# Create a batch request content
+batch_request_content = {
+    batch_request_item1.id: batch_request_item1,
+    batch_request_item2.id: batch_request_item2
+}
+
+batch_content = BatchRequestContent(batch_request_content)
+
+
+# Function to demonstrate the usage of BatchRequestBuilder
+async def main():
+    batch_response_content = await batch_request_builder.post(batch_request_content=batch_content)
+    # response_type=User
+
+    try:
+        individual_response = batch_response_content.get_response_by_id(
+            batch_request_item1.id, User
+        )
+        print(f"Individual Response: {individual_response}")
+    except AttributeError as e:
+        print(f"Error getting response by ID: {e}")
+
+
+asyncio.run(main())

samples/batch_requests/batch_response_get_status_codes.py

@@ -0,0 +1,25 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+#pylint: disable=undefined-variable
+"""Demonstrates getting status codes in Batch Responses"""
+batch_request_content = {
+    batch_request_item1.id: batch_request_item1,
+    batch_request_item2.id: batch_request_item2
+}
+
+batch_content = BatchRequestContent(batch_request_content)
+
+
+async def main():
+    batch_response_content = await batch_request_builder.post(batch_request_content=batch_content)
+
+    try:
+        status_codes = batch_response_content.get_response_status_codes()
+        print(f"Status Codes: {status_codes}")
+    except AttributeError as e:
+        print(f"Error getting respons status codes: {e}")
+
+
+asyncio.run(main())