JiriUhlir
106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
import time
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi import HTTPException, status
|
|
|
|
from .config import Settings
|
|
|
|
|
|
class MicrosoftGraphClient:
|
|
def __init__(self, settings: Settings) -> None:
|
|
self._settings = settings
|
|
self._access_token: str | None = None
|
|
self._expires_at = 0.0
|
|
|
|
async def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
params: dict[str, Any] | None = None,
|
|
json: Any | None = None,
|
|
content: bytes | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
) -> Any:
|
|
token = await self._get_access_token()
|
|
request_headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Accept": "application/json",
|
|
}
|
|
if headers:
|
|
request_headers.update(headers)
|
|
|
|
url = self._build_url(path)
|
|
async with httpx.AsyncClient(timeout=self._settings.request_timeout_seconds) as client:
|
|
response = await client.request(
|
|
method,
|
|
url,
|
|
params=params,
|
|
json=json,
|
|
content=content,
|
|
headers=request_headers,
|
|
)
|
|
|
|
if response.status_code >= 400:
|
|
raise self._graph_exception(response)
|
|
if response.status_code == status.HTTP_204_NO_CONTENT or not response.content:
|
|
return None
|
|
return response.json()
|
|
|
|
async def _get_access_token(self) -> str:
|
|
if self._access_token and time.time() < self._expires_at - 60:
|
|
return self._access_token
|
|
|
|
if not self._settings.is_configured:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail="Microsoft 365 credentials are not configured.",
|
|
)
|
|
|
|
data = {
|
|
"client_id": self._settings.client_id,
|
|
"client_secret": self._settings.client_secret,
|
|
"scope": self._settings.graph_scope,
|
|
"grant_type": "client_credentials",
|
|
}
|
|
async with httpx.AsyncClient(timeout=self._settings.request_timeout_seconds) as client:
|
|
response = await client.post(self._settings.token_url, data=data)
|
|
|
|
if response.status_code >= 400:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail={
|
|
"message": "Unable to authenticate with Microsoft identity platform.",
|
|
"upstream_status": response.status_code,
|
|
"upstream_response": self._safe_json(response),
|
|
},
|
|
)
|
|
|
|
payload = response.json()
|
|
self._access_token = payload["access_token"]
|
|
self._expires_at = time.time() + int(payload.get("expires_in", 3599))
|
|
return self._access_token
|
|
|
|
def _build_url(self, path: str) -> str:
|
|
if path.startswith("https://"):
|
|
return path
|
|
return f"{self._settings.graph_base_url.rstrip('/')}/{path.lstrip('/')}"
|
|
|
|
def _graph_exception(self, response: httpx.Response) -> HTTPException:
|
|
return HTTPException(
|
|
status_code=status.HTTP_502_BAD_GATEWAY,
|
|
detail={
|
|
"message": "Microsoft Graph request failed.",
|
|
"upstream_status": response.status_code,
|
|
"upstream_response": self._safe_json(response),
|
|
},
|
|
)
|
|
|
|
@staticmethod
|
|
def _safe_json(response: httpx.Response) -> Any:
|
|
try:
|
|
return response.json()
|
|
except ValueError:
|
|
return response.text
|