146 lines
6.2 KiB
Python
146 lines
6.2 KiB
Python
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
|