import requests
from msal import PublicClientApplication
from dotenv import load_dotenv
import os
# Load environment variables
load_dotenv()
# Configuration from environment variables
CLIENT_ID = os.getenv('CLIENT_ID')
# TENANT_ID is not used since we're using the 'consumers' endpoint
CLIENT_SECRET = os.getenv('CLIENT_SECRET') # This might not be necessary for public clients
USER_ID = os.getenv('USER_ID')
authority_url = 'https://login.microsoftonline.com/consumers'
scopes = ['Files.ReadWrite.All'] # Scope 'https://graph.microsoft.com/.default' might not be needed
app = PublicClientApplication(CLIENT_ID, authority=authority_url)
# The following is a simplistic approach to illustrate the flow
# You need a URL to redirect the user to for login, including the redirect URI
redirect_uri = 'http://localhost:8000/callback' # Make sure this matches the redirect URI configured in Azure
url = app.get_authorization_request_url(scopes, redirect_uri=redirect_uri)
print("Please go to this URL and sign-in:", url)
# After sign-in, you will receive a callback to the redirect URI with a code
# This part typically happens on your web server which handles the redirect
code = input("Enter the code you received: ")
result = app.acquire_token_by_authorization_code(code, scopes=scopes, redirect_uri=redirect_uri)
if 'access_token' in result:
access_token = result['access_token']
else:
print(result.get('error'))
print(result.get('error_description'))
def upload_file(access_token, file_path, destination_path):
headers = {
'Authorization': 'Bearer ' + access_token,
'Content-Type': 'application/octet-stream'
}
with open(file_path, 'rb') as file_data:
response = requests.put(
f'https://graph.microsoft.com/v1.0/users/{USER_ID}/drive/root:/{destination_path}:/content',
headers=headers,
data=file_data
)
return response.json()
У меня есть личная учетная запись Microsoft, и этот код использует API-интерфейс Microsoft GRAPH для загрузки моих изображений и ZIP-файлов размером не менее 2 ГБ на один диск. Однако то, что я пытаюсь сделать, я никогда не получаю токен доступа для загрузки. Я не знаю, чего мне не хватает.
Чтобы загрузить большие файлы в OneDrive с помощью API Microsoft Graph в Python, используйте вызов API createUploadSession.
Первоначально я зарегистрировал одно мультитенантное приложение, выбрав ниже поддерживаемый тип учетной записи:
Обязательно включите опцию общедоступных клиентских потоков и добавьте URI перенаправления на платформе Mobile and desktop applications
, как показано ниже:
В моем случае я запустил приведенный ниже модифицированный код Python для загрузки большого файла в OneDrive с помощью API Microsoft Graph:
import requests
from msal import PublicClientApplication
from dotenv import load_dotenv
import os
import time
load_dotenv()
# Configuration from environment variables
CLIENT_ID = os.getenv('CLIENT_ID')
CLIENT_SECRET = os.getenv('CLIENT_SECRET')
USER_ID = os.getenv('USER_ID')
authority_url = 'https://login.microsoftonline.com/consumers'
scopes = ['Files.ReadWrite.All']
app = PublicClientApplication(CLIENT_ID, authority=authority_url)
# Simplistic approach to illustrate the flow
redirect_uri = 'http://localhost:8000/callback'
url = app.get_authorization_request_url(scopes, redirect_uri=redirect_uri)
print("Please go to this URL and sign-in:", url)
# After sign-in, receive a callback with a code
code = input("Enter the code you received: ")
result = app.acquire_token_by_authorization_code(code, scopes=scopes, redirect_uri=redirect_uri)
if 'access_token' in result:
access_token = result['access_token']
print("Access token acquired.")
else:
print(result.get('error'))
print(result.get('error_description'))
exit(1)
def create_upload_session(access_token, file_name):
url = f'https://graph.microsoft.com/v1.0/users/{USER_ID}/drive/root:/{file_name}:/createUploadSession'
headers = {
'Authorization': 'Bearer ' + access_token,
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers)
response.raise_for_status()
return response.json()
def upload_file_in_chunks(upload_url, file_path, chunk_size=327680, max_retries=5):
with open(file_path, 'rb') as file:
file_size = os.path.getsize(file_path)
for i in range(0, file_size, chunk_size):
chunk_data = file.read(chunk_size)
headers = {
'Content-Length': str(len(chunk_data)),
'Content-Range': f'bytes {i}-{i + len(chunk_data) - 1}/{file_size}'
}
retries = 0
while retries < max_retries:
try:
response = requests.put(upload_url, headers=headers, data=chunk_data, timeout=60)
response.raise_for_status()
uploaded = min(i + len(chunk_data), file_size)
print(f'Uploaded {uploaded} out of {file_size} bytes ({uploaded * 100 / file_size:.2f}%)')
break
except (requests.exceptions.ConnectTimeout, requests.exceptions.ConnectionError) as e:
retries += 1
print(f"Retrying ({retries}/{max_retries}) due to: {e}")
time.sleep(2 ** retries) # Exponential backoff
if retries == max_retries:
raise
file_path = 'C:\\Users\\xxxxx\\Downloads\\demofolder.zip'
file_name = 'DemoFolder.zip'
# Create upload session
upload_session = create_upload_session(access_token, file_name)
upload_url = upload_session['uploadUrl']
print("Upload session created.")
# Upload the file in chunks
upload_file_in_chunks(upload_url, file_path)
print("File uploaded successfully.")
Когда вы запустите приведенный выше код, вы получите URL-адрес авторизации в консоли, который необходимо запустить в браузере:
При запуске вышеуказанного URL-адреса в браузере вам будет предложено войти в систему с помощью личной учетной записи Microsoft и после успешной аутентификации будет указано значение code
в адресной строке:
Теперь скопируйте это значение code
и вставьте его в консоль, которая сгенерирует токен и загрузит файл частями, как показано ниже:
Когда я проверил то же самое через некоторое время, я получил ответ ниже: «Файл загружен успешно»:
Чтобы убедиться в этом, я проверил портал OneDrive, где zip-файл успешно загружен, как показано ниже:
Обратите внимание, что нет необходимости настраивать сервер для получения ответов от localhost:8000/callback. Попытка сделать это может привести к сбою кода. После перехода туда, даже если нет соединения с локальным хостом, копирование кода непосредственно из адресной строки должно работать.