import requests import os from typing import List class SharePointGraphClient: def __init__(self, tenant_id, client_id, client_secret, host_name, site_name ): self.tenant_id = tenant_id self.client_id = client_id self.client_secret = client_secret self.base_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token" self.host_name = host_name self.site_name = site_name self.document_library = None self.root_folder = None self.access_token = None self.site_id = None self.drive_id = None self.__retrieve_access_token__() self.__retrieve_site_id__() def __retrieve_access_token__(self) -> None: # Body for the access token request body = { 'grant_type': 'client_credentials', 'client_id': self.client_id, 'client_secret': self.client_secret, 'scope': 'https://graph.microsoft.com/.default' } response = requests.post(self.base_url, headers={'Content-Type': 'application/x-www-form-urlencoded'}, data=body) self.access_token = response.json().get('access_token') # Extract access token from the response def __retrieve_site_id__(self) -> None: # Build URL to request site ID full_url = f'https://graph.microsoft.com/v1.0/sites/{self.host_name}:/sites/{self.site_name}?$select=id' response = requests.get(full_url, headers={'Authorization': f'Bearer {self.access_token}'}) self.site_id = response.json().get('id') # Return the site ID def __retrieve_drive_id__(self) -> None: # Retrieve drive IDs and names associated with a site drives_url = f'https://graph.microsoft.com/v1.0/sites/{self.site_id}/drives' response = requests.get(drives_url, headers={'Authorization': f'Bearer {self.access_token}'}) drives = response.json().get('value', []) for drive in drives: if drive['name'] == self.document_library: self.drive_id = drive['id'] def set_document_scope(self, document_library: str, root_folder: str = "") -> None: self.document_library = document_library self.root_folder = root_folder self.__retrieve_drive_id__() def get_site_permission(self): # Go to https://entra.microsoft.com/ and grant the following API permission # Microsoft Graph # Sites.Read.All # Files.Read.All # SharePoint # Sites.FullControl.All # Retrieve drive IDs and names associated with a site permission_url = f'https://graph.microsoft.com/v1.0/sites/{self.site_id}/permissions' response = requests.get(permission_url, headers={ 'Authorization': f'Bearer {self.access_token}' } ) print(response.json()) def create_site_permission(self, permisssions: List[str]): # Go to https://entra.microsoft.com/ and grant the following API permission # Microsoft Graph # Sites.Read.All # Files.Read.All # SharePoint # Sites.FullControl.All # Retrieve drive IDs and names associated with a site permission_url = f'https://graph.microsoft.com/v1.0/sites/{self.site_id}/permissions' body = { "roles": permisssions, "grantedToIdentities": [{ "application": { "id": self.client_id, "displayName": "sharepoint-media-content-read" } }] } response = requests.post(permission_url, headers={ 'Content-Type': 'application/json', 'Authorization': f'Bearer {self.access_token}' }, json=body ) print(response.json()) def get_item_id(self, relative_path: str) -> str: # Get the contents of a folder folder_url = f'https://graph.microsoft.com/v1.0/sites/{self.site_id}/drives/{self.drive_id}/root:/{self.root_folder.strip("/")+"/"+relative_path}' response = requests.get(folder_url, headers={'Authorization': f'Bearer {self.access_token}'}) items_data = response.json() return items_data['id'] if 'id' in items_data else None def list_folder_children(self, folder_relative_path): folder_id = self.get_item_id(relative_path=folder_relative_path) folder_contents_url = f'https://graph.microsoft.com/v1.0/sites/{self.site_id}/drives/{self.drive_id}/items/{folder_id}/children' contents_headers = {'Authorization': f'Bearer {self.access_token}'} contents_response = requests.get(folder_contents_url, headers=contents_headers) folder_contents = contents_response.json() items_list = [] # List to store information if 'value' in folder_contents: for item in folder_contents['value']: items_list.append( { "type": 'folder' if 'folder' in item else "file" if 'file' in item else "unknown", "name": item['name'], "id": item['id'] } ) return items_list def list_sub_folders(self, folder_relative_path): items_list = self.list_folder_children(folder_relative_path) return [item for item in items_list if item['type'] == 'folder'] def list_files(self, folder_relative_path): items_list = self.list_folder_children(folder_relative_path) return [item for item in items_list if item['type'] == 'file'] def get_file_content(self, file_id): file_url = f"https://graph.microsoft.com/v1.0/sites/{self.site_id}/drives/{self.drive_id}/items/{file_id}/content" headers = {'Authorization': f'Bearer {self.access_token}'} response = requests.get(file_url, headers=headers) return response.content