forked from ryan-blunden/django-chatgpt-clone
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathgraphapi.py
159 lines (121 loc) · 5.56 KB
/
graphapi.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
import os
import asyncio
from enum import IntEnum
from dotenv import load_dotenv
from azure.identity import DeviceCodeCredential , UsernamePasswordCredential
from azure.identity import ClientSecretCredential
from msgraph.generated.models.o_data_errors.o_data_error import ODataError
from msgraph import GraphServiceClient
from msgraph.generated.sites.sites_request_builder import SitesRequestBuilder
load_dotenv()
# class Graph:
# client_credential: ClientSecretCredential
# app_client: GraphServiceClient
# def __init__(self):
# tenant_id = os.environ['MS_GRAPH_TENANT_ID']
# client_id = os.environ['MS_GRAPH_APP_ID']
# client_secret = os.environ['MS_GRAPH_SECRET']
# self.client_credential = ClientSecretCredential(tenant_id, client_id, client_secret)
# self.app_client = GraphServiceClient(self.client_credential) # type: ignore
# async def get_app_only_token(self):
# graph_scope = 'https://graph.microsoft.com/.default'
# access_token = await self.client_credential.get_token(graph_scope)
# return access_token.token
# async def get_sites(self):
# query_params = SitesRequestBuilder.SitesRequestBuilderGetQueryParameters(
# # Only request specific properties
# select = ['displayName', 'id', 'name','lastModifiedDateTime'],
# # Get at most 25 results
# top = 25,
# # Sort by display name
# orderby= ['displayName']
# )
# request_config = SitesRequestBuilder.SitesRequestBuilderGetRequestConfiguration(
# query_parameters=query_params
# )
# sites = await self.app_client.sites.get(request_configuration=request_config)
# return sites
# async def display_access_token(graph: Graph):
# token = await graph.get_app_only_token()
# print('App-only token:', token, '\n')
# async def list_sites(graph: Graph):
# sites_page = await graph.get_sites()
# # Output each users's details
# if sites_page and sites_page.value:
# for site in sites_page.value:
# print('Site:', site.display_name)
# print(' ID:', site.id)
# print(' modified:', site.last_modified_date_time)
# # If @odata.nextLink is present
# more_available = sites_page.odata_next_link is not None
# print('\nMore sites available?', more_available, '\n')
# cf. https://learn.microsoft.com/en-us/graph/sdks/choose-authentication-providers?tabs=python#on-behalf-of-provider
class MSGraphMode(IntEnum):
APP_SECRET = 1
APP_USER = 2
mode : MSGraphMode = MSGraphMode.APP_SECRET
async def run_ms_graph():
# Multi-tenant apps can use "common",
# single-tenant apps must use the tenant ID from the Azure portal
tenant_id = os.environ['MS_GRAPH_TENANT_ID']
# Values from app registration
client_id = os.environ['MS_GRAPH_APP_ID']
# client secret
client_secret = os.environ['MS_GRAPH_SECRET']
site_id = os.environ['MS_GRAPH_SITE_ID']
if mode == MSGraphMode.APP_SECRET:
# The client credentials flow requires that you request the
# /.default scope, and pre-configure your permissions on the
# app registration in Azure. An administrator must grant consent
# to those permissions beforehand.
scopes = ['https://graph.microsoft.com/.default']
# azure.identity
# credential = DeviceCodeCredential(
# tenant_id=tenant_id,
# client_id=client_id)
# azure.identity.aio
credential = ClientSecretCredential(
tenant_id=tenant_id,
client_id=client_id,
client_secret=client_secret)
graph_client : GraphServiceClient = GraphServiceClient(credential, scopes)
print(credential.get_token('https://graph.microsoft.com/.default'))
sites = await graph_client.sites.get_all_sites.get()
print(sites)
for site in sites.value:
print(f"{site.id}: {site.name} / {site.display_name}")
site = await graph_client.sites.by_site_id(site_id).get()
print(site)
drives = await graph_client.drives.get()
print(drives)
drives2 = await graph_client.sites.by_site_id(site_id).drives.get()
print(drives2)
drive_id = drives2.value[0].id
#await graph_client.sites.by_site_id(site_id).drives.by_drive_id(drive_id).items
children = await graph_client.drives.by_drive_id(drive_id).items.by_drive_item_id('root').children.get()
child = children.value[0]
print(child.id)
print(child.name)
childs_children = await graph_client.drives.by_drive_id(drive_id).items.by_drive_item_id(child.id).children.get()
print(childs_children.value)
# shares = await graph_client.shares.get()
# print(shares)
# graph: Graph = Graph()
# await display_access_token(graph)
# await list_sites(graph)
elif mode == MSGraphMode.APP_USER:
scopes = ['User.Read']
# user on behalf of
username = os.environ['MS_GRAPH_USER_NAME']
password = os.environ['MS_GRAPH_USER_PWD']
credential = UsernamePasswordCredential(
tenant_id=tenant_id,
client_id=client_id,
username=username,
password=password)
graph_client : GraphServiceClient = GraphServiceClient(credential, scopes)
drives = await graph_client.me.drives.get()
print(drives)
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(run_ms_graph())