Coverage for mindsdb / integrations / handlers / sharepoint_handler / sharepoint_api.py: 0%
165 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-21 00:36 +0000
1import ast
2from datetime import datetime, timezone
3from typing import Text, List, Dict, Any
5from mindsdb.integrations.handlers.sharepoint_handler.utils import (
6 bearer_token_request,
7 get_an_entity,
8 delete_an_entity,
9 update_an_entity,
10 create_an_entity,
11)
14class SharepointAPI:
15 def __init__(
16 self, client_id: str = None, client_secret: str = None, tenant_id: str = None
17 ):
18 self.client_id = client_id
19 self.client_secret = client_secret
20 self.tenant_id = tenant_id
21 self.bearer_token = None
22 self.is_connected = False
23 self.expiration_time = datetime.now(timezone.utc).timestamp()
25 def get_bearer_token(self) -> None:
26 """
27 Generates new bearer token for the credentials
29 Returns
30 None
31 """
32 response = bearer_token_request(
33 client_id=self.client_id,
34 tenant_id=self.tenant_id,
35 client_secret=self.client_secret,
36 )
37 self.bearer_token = response["access_token"]
38 self.expiration_time = int(response["expires_on"])
39 self.is_connected = True
41 def check_bearer_token_validity(self) -> bool:
42 """
43 Provides information whether a valid bearer token is available or not. Returns true if available
44 otherwise false
46 Returns
47 bool
48 """
49 if (
50 self.is_connected
51 and datetime.now(timezone.utc).astimezone().timestamp()
52 < self.expiration_time
53 ):
54 return True
55 else:
56 return False
58 def disconnect(self) -> None:
59 """
60 Removes bearer token from the sharepoint API class (makes it null)
62 Returns
63 None
64 """
65 self.bearer_token = None
66 self.is_connected = False
68 def get_all_sites(self, limit: int = None) -> List[Dict[Text, Any]]:
69 """
70 Gets all sites associated with the account
72 limit: limits the number of site information to be returned
74 Returns
75 response: metadata information corresponding to all sites
76 """
77 url = "https://graph.microsoft.com/v1.0/sites?search=*"
78 response = get_an_entity(url=url, bearer_token=self.bearer_token)
79 if limit:
80 response = response[:limit]
81 return response
83 def update_sites(
84 self, site_dict: List[Dict[Text, Text]], values_to_update: Dict[Text, Any]
85 ) -> None:
86 """
87 Updates the given sites (site_dict) with the provided values (values_to_update)
88 Calls the function update_a_site for every site
89 site_dict: A dictionary containing site ids of the sites which are to be updated
90 values_to_update: a dictionary which will be used to update the fields of the sites
92 Returns
93 None
94 """
95 for site_entry in site_dict:
96 self.update_a_site(
97 site_id=site_entry["siteId"],
98 values_to_update=values_to_update,
99 )
101 def update_a_site(self, site_id: str, values_to_update: Dict[Text, Any]) -> None:
102 """
103 Updates a site with given values
104 site_id: GUID of the site
105 values_to_update: a dictionary values which will be used to update the properties of the site
107 Returns
108 None
109 """
110 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/"
111 update_an_entity(
112 url=url, values_to_update=values_to_update, bearer_token=self.bearer_token
113 )
115 def get_lists_by_site(
116 self, site_id: str, limit: int = None
117 ) -> List[Dict[Text, Any]]:
118 """
119 Gets lists' information corresponding to a site
121 site_id: GUID of a site
122 limit: limits the number of lists for which information is returned
124 Returns
125 response: metadata information/ fields corresponding to lists of the site
126 """
127 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/lists"
128 response = get_an_entity(url=url, bearer_token=self.bearer_token)
129 if limit:
130 response = response[:limit]
131 return response
133 def get_all_lists(self, limit: int = None) -> List[Dict[Text, Any]]:
134 """
135 Gets all the lists' information assocaited with the account
137 limit: puts a limit to the number of lists returned
139 Returns
140 response: returns metadata information regarding all the lists that have been made using that account
141 """
142 sites = self.get_all_sites()
143 lists = []
144 for site in sites:
145 for list_dict in self.get_lists_by_site(site_id=site["id"].split(",")[1]):
146 list_dict["siteName"] = site["name"]
147 list_dict["siteId"] = site["id"].split(",")[1]
148 lists.append(list_dict)
149 if limit:
150 lists = lists[:limit]
151 return lists
153 def delete_lists(self, list_dict: List[Dict[Text, Any]]) -> None:
154 """
155 Deletes lists for the given site ID and list ID
157 list_dict: a dictionary values containing the list IDs which are to be deleted and
158 their corresponding site IDs
160 Returns
161 None
162 """
163 for list_entry in list_dict:
164 self.delete_a_list(site_id=list_entry["siteId"], list_id=list_entry["id"])
166 def delete_a_list(self, site_id: str, list_id: str) -> None:
167 """
168 Deletes a list, given its list ID and its site ID
170 site_id: GUID of the site in which the list is present
171 list_id: GUID of the list which is to be deleted
173 Returns
174 None
175 """
176 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/lists/{list_id}"
177 delete_an_entity(url=url, bearer_token=self.bearer_token)
179 def update_lists(
180 self, list_dict: List[Dict[Text, Text]], values_to_update: Dict[Text, Any]
181 ) -> None:
182 """
183 Updates the given lists (list_dict) with the provided values (values_to_update)
184 Calls the function update_a_list for every list
185 list_dict: A dictionary containing ids of the list which are to be updated and also their site IDs
186 values_to_update: a dictionary which will be used to update the fields of the lists
188 Returns
189 None
190 """
191 for list_entry in list_dict:
192 self.update_a_list(
193 site_id=list_entry["siteId"],
194 list_id=list_entry["id"],
195 values_to_update=values_to_update,
196 )
198 def update_a_list(
199 self, site_id: str, list_id: str, values_to_update: Dict[Text, Any]
200 ) -> None:
201 """
202 Updates a list with given values
203 list_id: GUID of the list
204 site_id: GUID of the site in which the list is present
205 values_to_update: a dictionary values which will be used to update the properties of the list
207 Returns
208 None
209 """
210 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/lists/{list_id}/"
211 update_an_entity(
212 url=url, bearer_token=self.bearer_token, values_to_update=values_to_update
213 )
215 def create_lists(self, data: List[Dict[Text, Any]]) -> None:
216 """
217 Creates lists with the information provided in the data parameter
218 calls create_a_list for each entry of list metadata dictionary
220 data: parameter which contains information such as the site IDs where the lists would be created
221 and their metadata information which will be used to create them
223 Returns
224 None
225 """
226 for entry in data:
227 self.create_a_list(
228 site_id=entry["siteId"],
229 column=entry.get("column"),
230 display_name=entry["displayName"],
231 list_template=entry["list"],
232 )
234 def create_a_list(
235 self, site_id: str, list_template: str, display_name: str, column: str = None
236 ) -> None:
237 """
238 Creates a list with metadata information provided in the params
240 site_id: GUID of the site where the list is to be created
241 list_template: a string which contains the list template information (type of list)
242 eg.- "{'template': 'documentLibrary'}"
243 display_name: the display name of the given list, which will be displayed in the site
244 column: specifies the list of columns that should be created for the list
245 eg.- "[{'name': 'Author', 'text': { }},{'name': 'PageCount', 'number': { }}]"
247 Returns
248 None
249 """
250 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/lists/"
251 payload = {}
252 if column:
253 column = ast.literal_eval(column)
254 payload["column"] = column
255 payload["displayName"] = display_name
256 payload["list"] = ast.literal_eval(list_template)
257 create_an_entity(url=url, payload=payload, bearer_token=self.bearer_token)
259 def get_site_columns_by_site(
260 self, site_id: str, limit: int = None
261 ) -> List[Dict[Text, Any]]:
262 """
263 Gets columns' information corresponding to a site
265 site_id: GUID of a site
266 limit: limits the number of columns for which information is returned
268 Returns
269 response: metadata information/ fields corresponding to columns of the site
270 """
271 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/columns/"
272 response = get_an_entity(url=url, bearer_token=self.bearer_token)
273 if limit:
274 response = response[:limit]
275 return response
277 def get_all_site_columns(self, limit: int = None) -> List[Dict[Text, Any]]:
278 """
279 Gets all the columns' information associated with the account
281 limit: puts a limit to the number of columns returned
283 Returns
284 response: returns metadata information regarding all the columns that have been made using that account
285 """
286 sites = self.get_all_sites()
287 site_columns = []
288 for site in sites:
289 for site_column_dict in self.get_site_columns_by_site(
290 site_id=site["id"].split(",")[1]
291 ):
292 site_column_dict["siteName"] = site["name"]
293 site_column_dict["siteId"] = site["id"].split(",")[1]
294 site_columns.append(site_column_dict)
295 if limit:
296 site_columns = site_columns[:limit]
297 return site_columns
299 def update_site_columns(
300 self,
301 site_column_dict: List[Dict[Text, Text]],
302 values_to_update: Dict[Text, Any],
303 ) -> None:
304 """
305 Updates the given columns (site_column_dict) with the provided values (values_to_update)
306 Calls the function update_a_site_column for every column
308 site_column_dict: A dictionary containing ids of the column which are to be updated and
309 also their site IDs
310 values_to_update: a dictionary which will be used to update the fields of the columns
312 Returns
313 None
314 """
315 for site_column_entry in site_column_dict:
316 self.update_a_site_column(
317 site_id=site_column_entry["siteId"],
318 column_id=site_column_entry["id"],
319 values_to_update=values_to_update,
320 )
322 def update_a_site_column(
323 self, site_id: str, column_id: str, values_to_update: Dict[Text, Any]
324 ):
325 """
326 Updates a column with given values
328 column_id: GUID of the column
329 site_id: GUID of the site in which the column is present
330 values_to_update: a dictionary values which will be used to update the properties of the column
332 Returns
333 None
334 """
335 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/columns/{column_id}"
336 update_an_entity(
337 url=url, values_to_update=values_to_update, bearer_token=self.bearer_token
338 )
340 def delete_site_columns(self, column_dict: List[Dict[Text, Any]]) -> None:
341 """
342 Deletes columns for the given site ID and column ID
344 column_dict: a dictionary values containing the column IDs which are to be deleted and
345 their corresponding site IDs
347 Returns
348 None
349 """
350 for column_entry in column_dict:
351 self.delete_a_site_columns(
352 site_id=column_entry["siteId"], column_id=column_entry["id"]
353 )
355 def delete_a_site_columns(self, site_id: str, column_id: str) -> None:
356 """
357 Deletes a column, given its column ID and its site ID
359 site_id: GUID of the site in which the column is present
360 column_id: GUID of the column which is to be deleted
362 Returns
363 None
364 """
365 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/columns/{column_id}"
366 delete_an_entity(url=url, bearer_token=self.bearer_token)
368 def create_site_columns(self, data: List[Dict[Text, Any]]) -> None:
369 """
370 Creates columns with the information provided in the data parameter
371 calls create_a_site_column for each entry of column metadata dictionary
373 data: parameter which contains information such as the site IDs where the columns would be created
374 and their metadata information which will be used to create them
376 Returns
377 None
378 """
379 for entry in data:
380 self.create_a_site_column(
381 site_id=entry["siteId"],
382 enforce_unique_values=entry.get("enforceUniqueValues"),
383 hidden=entry.get("hidden"),
384 indexed=entry.get("indexed"),
385 name=entry["name"],
386 text=entry.get("text"),
387 )
389 def create_a_site_column(
390 self,
391 site_id: str,
392 enforce_unique_values: bool,
393 hidden: bool,
394 indexed: bool,
395 name: str,
396 text: str = None,
397 ) -> None:
398 """
399 Creates a list with metadata information provided in the params
401 site_id: GUID of the site where the column is to be created
402 enforced_unique_values: if true, no two list items may have the same value for this column
403 hidden: specifies whether the column is displayed in the user interface
404 name: the API-facing name of the column as it appears in the fields on a listItem.
405 text: details regarding the text values in the column
407 Returns
408 None
409 """
411 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/columns/"
412 payload = {}
413 if text:
414 text = ast.literal_eval(text)
415 payload["text"] = text
416 payload["name"] = name
417 if enforce_unique_values is not None:
418 payload["enforceUniqueValues"] = enforce_unique_values
419 if hidden is not None:
420 payload["hidden"] = hidden
421 if indexed is not None:
422 payload["indexed"] = indexed
423 create_an_entity(url=url, payload=payload, bearer_token=self.bearer_token)
425 def get_items_by_sites_and_lists(
426 self, site_id: str, list_id: str, limit: int = None
427 ) -> List[Dict[Text, Any]]:
428 """
429 Gets items' information corresponding to a site and a list
431 site_id: GUID of a site
432 list_id: GUID of a list
433 limit: limits the number of columns for which information is returned
435 Returns
436 response: metadata information/ fields corresponding to list-items of the site
437 """
438 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/lists/{list_id}/items?expand=fields&select=*"
439 response = get_an_entity(url=url, bearer_token=self.bearer_token)
440 if limit:
441 response = response[:limit]
442 return response
444 def get_all_items(self, limit: int = None) -> List[Dict[Text, Any]]:
445 """
446 Gets all the items' information associated with the account
448 limit: puts a limit to the number of items returned
450 Returns
451 response: returns metadata information regarding all the items that are associated with that account
452 """
453 sites = self.get_all_sites()
454 items = []
455 for site in sites:
456 site_id = site["id"].split(",")[1]
457 for sharepoint_list in self.get_lists_by_site(site_id=site_id):
458 for item_dict in self.get_items_by_sites_and_lists(
459 site_id=site["id"].split(",")[1], list_id=sharepoint_list["id"]
460 ):
461 item_dict["siteName"] = site["name"]
462 item_dict["siteId"] = site["id"].split(",")[1]
463 item_dict["listId"] = sharepoint_list["id"]
464 item_dict["list_name"] = sharepoint_list["displayName"]
465 items.append(item_dict)
466 if limit:
467 items = items[:limit]
468 return items
470 def update_items(
471 self, item_dict: List[Dict[Text, Text]], values_to_update: Dict[Text, Any]
472 ) -> None:
473 """
474 Updates the given items (item_dict) with the provided values (values_to_update)
475 Calls the function update_a_item for every column
477 item_dict: A dictionary containing ids of the list-items which are to be updated and
478 also their site IDs
479 values_to_update: a dictionary which will be used to update the fields of the items
481 Returns
482 None
483 """
484 for item_entry in item_dict:
485 self.update_an_item(
486 site_id=item_entry["siteId"],
487 list_id=item_entry["listId"],
488 item_id=item_entry["id"],
489 values_to_update=values_to_update,
490 )
492 def update_an_item(
493 self,
494 site_id: str,
495 list_id: str,
496 item_id: str,
497 values_to_update: Dict[Text, Any],
498 ):
499 """
500 Updates an item with given values
502 item_id: GUID of the column
503 list_id: GUID of the list
504 site_id: GUID of the site in which the list is present
505 values_to_update: a dictionary values which will be used to update the properties of the list-item
507 Returns
508 None
509 """
510 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/lists/{list_id}/items/{item_id}"
511 update_an_entity(
512 url=url, values_to_update=values_to_update, bearer_token=self.bearer_token
513 )
515 def delete_items(self, item_dict: List[Dict[Text, Any]]) -> None:
516 """
517 Deletes items for the given site ID and list ID
519 item_dict: a dictionary values containing the item IDs which are to be deleted,
520 their corresponding site IDs and their list IDs
522 Returns
523 None
524 """
525 for item_entry in item_dict:
526 self.delete_an_item(
527 site_id=item_entry["siteId"],
528 list_id=item_entry["listId"],
529 item_id=item_entry["id"],
530 )
532 def delete_an_item(self, site_id: str, list_id: str, item_id: str) -> None:
533 """
534 Deletes an item, given its item ID, its site ID and its list ID
536 list_id: GUID of the list in which the site is present
537 site_id: GUID of the site in which the list is present
538 item_id: GUID of the item which is to be deleted
540 Returns
541 None
542 """
543 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/lists/{list_id}/items/{item_id}"
544 delete_an_entity(url=url, bearer_token=self.bearer_token)
546 def create_items(self, data: List[Dict[Text, Any]]) -> None:
547 """
548 Creates items with the information provided in the data parameter
549 calls create_an_item for each entry of item metadata dictionary
551 data: parameter which contains information such as the site IDs and list IDs where the items
552 would be created and their metadata information which will be used to create them
554 Returns
555 None
556 """
557 for entry in data:
558 self.create_an_item(
559 site_id=entry["siteId"],
560 list_id=entry["listId"],
561 fields=entry.get("fields"),
562 )
564 def create_an_item(self, site_id: str, list_id: str, fields: str) -> None:
565 """
566 Creates an item with metadata information provided in the params
568 site_id: GUID of the site where the list id present
569 list_id: GUID of the list where the item is to be created
570 fields: The values of the columns set on this list item.
572 Returns
573 None
574 """
575 url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/lists/{list_id}/items/"
576 payload = {}
577 if fields:
578 payload["fields"] = ast.literal_eval(fields)
579 create_an_entity(url=url, payload=payload, bearer_token=self.bearer_token)