I'm encountering an issue while integrating my application with the Microsoft Graph API. Below, I'm providing the relevant code snippets along with the logs for the application's behavior
Code
import logging
from fastapi import FastAPI, Depends, HTTPException, status, Request, Response
from fastapi.security import OAuth2AuthorizationCodeBearer
from starlette.responses import RedirectResponse
from starlette.templating import Jinja2Templates
import httpx
from urllib.parse import quote
import os
# Configure logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
app = FastAPI()
templates = Jinja2Templates(directory="templates")
# Replace these with your Microsoft Azure Application details
CLIENT_ID = ""
CLIENT_SECRET = ""
REDIRECT_URI = "http://localhost:5009/callback"
AUTHORITY = "https://login.microsoftonline.com/common"
SCOPE = "Calendars.Read.Shared"
AUTHORIZATION_URL = f"{AUTHORITY}/oauth2/v2.0/authorize"
TOKEN_URL = f"{AUTHORITY}/oauth2/v2.0/token"
# Define the authorization URL separately
authorization_url = (
f"{AUTHORIZATION_URL}?client_id={CLIENT_ID}"
f"&response_type=code&redirect_uri={quote(REDIRECT_URI)}"
f"&response_mode=query&scope={quote(SCOPE)}"
)
oauth2_scheme = OAuth2AuthorizationCodeBearer(
authorizationUrl=authorization_url,
tokenUrl=TOKEN_URL,
)
# Placeholder for storing access tokens
access_tokens = {}
@app.get("/")
async def landing_page(request: Request):
return templates.TemplateResponse("landing.html", {"request": request})
@app.get("/login")
async def login():
return RedirectResponse(url=authorization_url)
async def get_token(code: str):
headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
data = {
"grant_type": "authorization_code",
"code": code,
"redirect_uri": REDIRECT_URI,
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"scope": SCOPE,
}
async with httpx.AsyncClient() as client:
response = await client.post(TOKEN_URL, headers=headers, data=data)
response.raise_for_status()
logging.debug("Token Response: %s", response.text)
token_data = response.json()
access_token = token_data.get("access_token")
if access_token:
logging.debug("Access Token Retrieved Successfully")
else:
logging.error("Failed to retrieve access token")
return access_token
@app.get("/callback")
async def callback(request: Request, code: str):
logging.debug("Callback received with code: %s", code)
token = await get_token(code)
if token:
return RedirectResponse(url="/calendar")
else:
return HTTPException(status_code=500, detail="Failed to retrieve access token")
@app.get("/calendar")
async def read_calendar(request: Request, access_token: str = Depends(oauth2_scheme)):
logging.info("Calendar api *** ")
headers = {
"Authorization": f"Bearer {access_token}"
}
async with httpx.AsyncClient() as client:
response = await client.get("https://graph.microsoft.com/Calendars.Read.Shared", headers=headers)
logging.debug("Calendar Response: %s", response)
response.raise_for_status()
calendar_data = response.json()
return templates.TemplateResponse("calendar.html", {"request": request, "calendar_data": calendar_data})
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=5009)
Logs
- The server starts successfully, and the application startup process completes.
- Uvicorn is running on http://127.0.0.1:5009, and it's waiting for incoming requests.
- Requests are made to the server:
- A GET request to "/" returns a "200 OK" response.
- A GET request to "/login" returns a "307 Temporary Redirect" response.
- A POST request is made to "https://login.microsoftonline.com/common/oauth2/v2.0/token", and a token is received.
- A callback is received with a code from Microsoft.
- Another POST request is made to the token endpoint, and another token is received.
- The server receives a GET request to "/callback" with a code, and a redirect is issued.
- A GET request to "/calendar" returns a "401 Unauthorized" response.
