Microsoft Graph Api UnknownError in batch request
It’s possible to optimize the usage of Microsoft Graph Api by batching requests. There is a limitation for 20 requests in one batch request.
Here is a simplified examle:
api_requests = {
1: {
'id': 1,
'url': '/myorganization/users/<user_id1>/getMemberGroups',
'method': 'GET',
},
2: {
'id': 2,
'url': '/myorganization/users/<user_id2>/getMemberGroups',
'method': 'GET'
}
}
def batch_requests(api_requests, access_token):
request_json = [{
"id": request['id'],
"url": request['url'],
"method": request.get('method', 'GET'),
"body": request.get('body'),
"headers": request.get('headers', {"Content-Type": "application/json"})
} for request in api_requests.values()]
request_json = {'requests': request_json}
response = requests.request(
'POST',
url='https://graph.microsoft.com/beta/$batch',
json=request_json,
headers={'Authorization': 'Bearer ' + access_token}).json()
if 'error' in response:
# process error
pass
results = {}
for response in response['responses']:
results[response['id']] = (response['status'], response['body'])
return results
There is a list of possible errors and recommendations how to work with them.
But there is no information about UnknownError, it arrives with status code 503 or 504 and it does not appear often. In my practice it’s enough just to repeat the failed requests.
def batch_requests(api_requests, attempt=0):
request_json = [{
"id": request['id'],
"url": request['url'],
"method": request.get('method', 'GET'),
"body": request.get('body'),
"headers": request.get('headers', {"Content-Type": "application/json"})
} for request in api_requests.values()]
request_json = {'requests': request_json}
response = requests.request(
'POST',
url='https://graph.microsoft.com/beta/$batch',
json=request_json,
headers={'Authorization': 'Bearer ' + access_token}).json()
if 'error' in response:
# process error
pass
failed_requests = {}
results = {}
for response in response['responses']:
if 'error' in response['body']:
if response['body']['error']['code'] == 'UnknownError':
failed_requests[response['id']] = api_requests[response['id']]
else:
results[response['id']] = (response['status'], response['body'])
if failed_requests and attempt == 0:
results.update(batch_requests(failed_requests, attempt + 1))
return results