Coverage for mindsdb / integrations / utilities / handlers / api_utilities / microsoft / ms_graph_api_utilities.py: 77%
56 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-21 00:36 +0000
1import requests
2import time
3from typing import Dict, Generator, List, Optional, Text, Union
5from mindsdb.utilities import log
7logger = log.getLogger(__name__)
10class MSGraphAPIBaseClient:
11 """
12 The base class for the Microsoft Graph API clients.
13 This class contains common methods for accessing the Microsoft Graph API.
15 Attributes:
16 MICROSOFT_GRAPH_BASE_API_URL (Text): The base URL of the Microsoft Graph API.
17 MICROSOFT_GRAPH_API_VERSION (Text): The version of the Microsoft Graph API.
18 PAGINATION_COUNT (Optional[int]): The number of items to retrieve per request.
19 """
20 MICROSOFT_GRAPH_BASE_API_URL: Text = "https://graph.microsoft.com/"
21 MICROSOFT_GRAPH_API_VERSION: Text = "v1.0"
22 PAGINATION_COUNT: Optional[int] = 20
24 def __init__(self, access_token: Text) -> None:
25 """
26 Initializes the Microsoft Graph API client.
28 Args:
29 access_token (Text): The access token for authenticating the requests to the Microsoft Graph API.
30 """
31 self.access_token = access_token
32 self._group_ids = None
34 def _get_api_url(self, endpoint: Text) -> Text:
35 """
36 Constructs the API URL for the specified endpoint.
38 Args:
39 endpoint (Text): The endpoint of the Microsoft Graph API.
41 Returns:
42 Text: The fully constructed API URL.
43 """
44 api_url = f"{self.MICROSOFT_GRAPH_BASE_API_URL}{self.MICROSOFT_GRAPH_API_VERSION}/{endpoint}/"
45 return api_url
47 def _make_request(
48 self,
49 api_url: Text,
50 params: Optional[Dict] = None,
51 data: Optional[Dict] = None,
52 method: Text = "GET"
53 ) -> Union[Dict, object]:
54 """
55 Makes a request to the Microsoft Graph API.
57 Args:
58 api_url (Text): The API URL to make the request to.
59 params (Optional[Dict]): The parameters to include in the request.
60 data (Optional[Dict]): The data to include in the request.
61 method (Text): The HTTP method to use for the request.
63 Returns:
64 Union[Dict, object]: The response content of the request.
65 """
66 headers = {"Authorization": f"Bearer {self.access_token}"}
68 # Make the request to the Microsoft Graph API based on the method.
69 if method == "GET": 69 ↛ 71line 69 didn't jump to line 71 because the condition on line 69 was always true
70 response = requests.get(api_url, headers=headers, params=params)
71 elif method == "POST":
72 response = requests.post(api_url, headers=headers, json=data)
73 else:
74 raise NotImplementedError(f"Method {method} not implemented")
76 # Process the response.
77 # If the response is a 429 (rate limit exceeded), wait for the specified time and retry.
78 if response.status_code == 429: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true
79 if "Retry-After" in response.headers:
80 pause_time = float(response.headers["Retry-After"])
81 time.sleep(pause_time)
82 response = requests.get(api_url, headers=headers, params=params)
84 # If the response is not successful, raise an exception.
85 if response.status_code not in [200, 201]:
86 raise requests.exceptions.RequestException(response.text)
88 return response
90 def fetch_paginated_data(self, endpoint: Text, params: Optional[Dict] = None) -> Generator:
91 """
92 Fetches data from the Microsoft Graph API by making the specified request and handling pagination.
94 Args:
95 endpoint (str): The endpoint of the Microsoft Graph API to fetch data from.
96 params (Optional[Dict]): The parameters to include in the request.
98 Yields:
99 List: The data fetched from the Microsoft Graph API.
100 """
101 if params is None: 101 ↛ 103line 101 didn't jump to line 103 because the condition on line 101 was always true
102 params = {}
103 api_url = self._get_api_url(endpoint)
105 # Add the pagination count to the request parameters.
106 if "$top" not in params: 106 ↛ 109line 106 didn't jump to line 109 because the condition on line 106 was always true
107 params["$top"] = self.PAGINATION_COUNT
109 while api_url:
110 # Make the initial request to the Microsoft Graph API.
111 response = self._make_request(api_url, params)
112 response_json = response.json()
113 value = response.json()["value"]
115 # Get the next page of data if pagination is enabled.
116 params = None
117 api_url = response_json.get("@odata.nextLink", "")
118 yield value
120 def _fetch_data(self, endpoint: str, params: Optional[Dict] = {}) -> Union[List, Dict, bytes]:
121 """
122 Fetches data from the Microsoft Graph API by making the specified request.
124 Args:
125 endpoint (str): The endpoint of the Microsoft Graph API to fetch data from.
126 params (Optional[Dict]): The parameters to include in the request.
128 Returns:
129 Union[List, Dict, bytes]: The data fetched from the Microsoft Graph API.
130 """
131 api_url = self._get_api_url(endpoint)
133 response = self._make_request(api_url, params)
134 return response
136 def fetch_data_content(self, endpoint: str, params: Optional[Dict] = {}) -> bytes:
137 """
138 Fetches data content from the Microsoft Graph API by making the specified request.
140 Args:
141 endpoint (str): The endpoint of the Microsoft Graph API to fetch data from.
142 params (Optional[Dict]): The parameters to include in the request.
144 Returns:
145 bytes: The data content fetched from the Microsoft Graph API.
146 """
147 response = self._fetch_data(endpoint, params)
148 return response.content
150 def fetch_data_json(self, endpoint: str, params: Optional[Dict] = {}) -> Union[List, Dict]:
151 """
152 Fetches data from the Microsoft Graph API by making the specified request and returns the JSON response.
154 Args:
155 endpoint (str): The endpoint of the Microsoft Graph API to fetch data from.
156 params (Optional[Dict]): The parameters to include in the request.
158 Returns:
159 Union[List, Dict]: The JSON response fetched from the Microsoft Graph API.
160 """
161 response = self._fetch_data(endpoint, params)
162 response_json = response.json()
164 if "value" in response_json:
165 return response_json["value"]
166 return response_json