Created
August 29, 2025 08:57
-
-
Save hidirbozkurt/81e7e04f339fe42d664f69ebe64e1c96 to your computer and use it in GitHub Desktop.
Accessing OneDrive Files with Python using the Graph API (Personal Microsoft accounts only - Delegated permissions)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # Import required libraries for auth, browser flow, local server, URL parsing, file/JSON ops, and HTTP. | |
| import msal | |
| import webbrowser | |
| from http.server import BaseHTTPRequestHandler, HTTPServer | |
| import urllib.parse as urlparse | |
| import os | |
| import json | |
| import requests | |
| # Define Azure app settings, redirect URI, and Graph scopes. | |
| CLIENT_ID = "API_client_id" | |
| CLIENT_SECRET = "API_client_secret" | |
| AUTHORITY = "https://login.microsoftonline.com/consumers" | |
| REDIRECT_URI = "http://localhost:8000/callback" | |
| SCOPES = ["Files.Read.All", "User.Read"] | |
| # Prepare a persistent token cache to avoid re-login on subsequent runs. | |
| CACHE_PATH = "msal_token_cache.bin" | |
| token_cache = msal.SerializableTokenCache() | |
| if os.path.exists(CACHE_PATH): | |
| with open(CACHE_PATH, "r", encoding="utf-8") as f: | |
| token_cache.deserialize(f.read()) | |
| # Helper to save the token cache back to disk only when it changed. | |
| def persist_cache_if_changed(): | |
| if token_cache.has_state_changed: | |
| with open(CACHE_PATH, "w", encoding="utf-8") as f: | |
| f.write(token_cache.serialize()) | |
| # Initialize the MSAL confidential client using client secret and attach the cache. | |
| app = msal.ConfidentialClientApplication( | |
| client_id=CLIENT_ID, | |
| authority=AUTHORITY, | |
| client_credential=CLIENT_SECRET, | |
| token_cache=token_cache | |
| ) | |
| # Build the authorization URL for interactive sign-in and consent. | |
| auth_url = app.get_authorization_request_url( | |
| SCOPES, | |
| redirect_uri=REDIRECT_URI, | |
| state="delegated_auth" | |
| ) | |
| # Open the user's default browser to start the interactive sign-in and consent flow. | |
| print("Opening browser for sign-in...") | |
| webbrowser.open(auth_url) | |
| # Minimal HTTP server handler: captures the authorization response on the redirect URI. | |
| class Handler(BaseHTTPRequestHandler): | |
| # Handle GET requests coming to the redirect URI (to read the auth code). | |
| def do_GET(self): | |
| # Parse the request URL and extract query parameters. | |
| parsed = urlparse.urlparse(self.path) | |
| qs = urlparse.parse_qs(parsed.query) | |
| # If an authorization code is present, exchange it for tokens. | |
| if "code" in qs: | |
| code = qs["code"][0] | |
| result = app.acquire_token_by_authorization_code( | |
| code=code, | |
| scopes=SCOPES, | |
| redirect_uri=REDIRECT_URI | |
| ) | |
| # On success: respond in the browser and persist tokens to cache. | |
| if "access_token" in result: | |
| self.send_response(200) | |
| self.end_headers() | |
| self.wfile.write(b"Authentication successful. You can close this tab.\n") | |
| persist_cache_if_changed() | |
| print("Token saved to cache. You can use acquire_token_silent() next time.") | |
| # On failure: notify the browser and log the error details. | |
| else: | |
| self.send_response(400) | |
| self.end_headers() | |
| self.wfile.write(b"Failed to obtain token.\n") | |
| print("Error:", json.dumps(result, indent=2)) | |
| # If no code is present: return a 400 with a simple message. | |
| else: | |
| self.send_response(400) | |
| self.end_headers() | |
| self.wfile.write(b"Missing 'code' query parameter.\n") | |
| # Start the one-shot local HTTP server to handle a single redirect request. | |
| server = HTTPServer(("localhost", 8000), Handler) | |
| server.handle_request() | |
| app = msal.ConfidentialClientApplication( | |
| client_id=CLIENT_ID, | |
| authority=AUTHORITY, | |
| client_credential=CLIENT_SECRET, | |
| token_cache=token_cache | |
| ) | |
| accounts = app.get_accounts() | |
| if not accounts: | |
| raise SystemExit("No cached account found. Run the login script first.") | |
| result = app.acquire_token_silent(scopes=SCOPES, account=accounts[0]) | |
| if not result: | |
| raise SystemExit("Silent token acquisition failed. Run the login script again.") | |
| access_token = result["access_token"] | |
| headers = {"Authorization": f"Bearer {access_token}"} | |
| drive_url = "https://graph.microsoft.com/v1.0/me/drive/root/children" | |
| resp = requests.get(drive_url, headers=headers) | |
| print("Status:", resp.status_code) | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| items = data.get("value", []) | |
| print(f"\n=== OneDrive ROOT ITEMS ({len(items)}) ===\n") | |
| for i, item in enumerate(items, 1): | |
| name = item.get("name", "N/A") | |
| size = item.get("size", 0) | |
| is_folder = "folder" in item | |
| item_type = "Folder" if is_folder else "File" | |
| web_url = item.get("webUrl", "N/A") | |
| print(f"{i}. [{item_type}] {name} | Size: {size} | URL: {web_url}") | |
| else: | |
| print(f"Drive API error: {resp.status_code}") | |
| try: | |
| print(json.dumps(resp.json(), ensure_ascii=False, indent=2)) | |
| except Exception: | |
| print(resp.text) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment