scripts/graph_client.py — Data Room Toolkit

#!/usr/bin/env python3
# Data Room Toolkit 0.1.5
# By Zero Width Lab (zerowidthlab.com)
"""Shared Microsoft Graph authentication, transport, and configuration helpers."""
from __future__ import annotations
import base64
import json
import mimetypes
import os
import re
import sys
import time
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Iterable, Mapping
from pathlib import Path
from typing import Any
__author__ = "Zero Width Lab"
GRAPH_BASE_URL = "https://graph.microsoft.com/v1.0"
LOGIN_BASE_URL = "https://login.microsoftonline.com"
GRAPH_SCOPE = "https://graph.microsoft.com/.default"
MAX_COLLECTION_PAGES = 1000
UPLOAD_CHUNK_UNIT = 327680
UPLOAD_REQUEST_LIMIT = 60 * 1024 * 1024
_ALLOWED_DESTINATION_HOSTS = frozenset(
  {"graph.microsoft.com", "login.microsoftonline.com"}
)
GRAPH_ISSUED_GRAPH_HOST = "graph.microsoft.com"
TENANT_CONTENT_HOST_PATTERN = r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.sharepoint\.com"
CONFIG_DEFAULTS: dict[str, Any] = {
  "root_path": "",
  "request_timeout_seconds": 60,
  "throttle_seconds": 0.5,
  "max_retries": 4,
  "upload_chunk_bytes": 5 * 1024 * 1024,
  "small_file_threshold_bytes": 4 * 1024 * 1024,
}
MIME_TYPE_OVERRIDES = {
  ".csv": "text/csv",
  ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
  ".json": "application/json",
  ".md": "text/markdown",
  ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
  ".txt": "text/plain",
  ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
}
class GraphClientError(RuntimeError):
  """A safe error raised before an HTTP result can be returned."""
  def __init__(
    self,
    message: str,
    *,
    status: int | None = None,
    payload: Any = None,
    headers: Mapping[str, str] | None = None,
  ) -> None:
    super().__init__(message)
    self.status = status
    self.payload = payload
    self.headers = dict(headers or {})
class WriteOutcomeUnknown(GraphClientError):
  """A write may have committed on the server but its result is unconfirmed."""
WRITE_OUTCOME_UNKNOWN = (
  "outcome unknown: the write may have completed; run the dry run again "
  "to see the current destination state"
)
class CredentialError(GraphClientError):
  """A required credential environment variable is absent."""
class ConfigError(GraphClientError):
  """The optional target configuration is missing or invalid."""
_fake_path: str | None = None
_fake_responses: list[dict[str, Any]] | None = None
_contacted_hosts: set[str] = set()
_FAKE_TRANSPORT_ACK_VAR = "GRAPH_FAKE_TRANSPORT_TEST_ACK"
_FAKE_TRANSPORT_ACK_VALUE = "zero-width-lab-test-suite-only"
def _fake_transport_enabled() -> bool:
  return (
    bool(os.environ.get("GRAPH_FAKE_RESPONSES"))
    and os.environ.get(_FAKE_TRANSPORT_ACK_VAR) == _FAKE_TRANSPORT_ACK_VALUE
  )
def contacted_hosts() -> list[str]:
  """Return the sorted destination hosts contacted during this process run."""
  return sorted(_contacted_hosts)
def reset_contacted_hosts() -> None:
  """Clear destination host tracking, primarily for CLI and test isolation."""
  _contacted_hosts.clear()
def is_graph_issued_host(hostname: str) -> bool:
  """Return true for Graph or a single-label tenant SharePoint content host."""
  host = hostname.lower()
  if host == GRAPH_ISSUED_GRAPH_HOST:
    return True
  return re.fullmatch(TENANT_CONTENT_HOST_PATTERN, host) is not None
def _guard_destination(url: str, *, graph_issued_url: bool) -> str:
  """Return an allowed destination host or fail before transport is invoked.
    Ordinary requests may reach only the fixed Graph and sign-in hosts. A
    Graph-issued URL (an upload-session URL or a content-download redirect) may
    reach only Graph or a tenant SharePoint content host. Every URL, including
    each redirect hop, must use HTTPS on the default port without user
    information.
    """
  parsed = urllib.parse.urlsplit(url)
  if parsed.scheme.lower() != "https":
    raise GraphClientError("request destination URL must use HTTPS")
  hostname = parsed.hostname
  if not hostname:
    raise GraphClientError("request destination URL does not contain a hostname")
  if "@" in parsed.netloc:
    raise GraphClientError(
      "request destination URL must not contain user information"
    )
  try:
    port = parsed.port
  except ValueError as exc:
    raise GraphClientError(
      "request destination URL contains an invalid port"
    ) from exc
  if port is not None and port != 443:
    raise GraphClientError(
      "request destination URL must use the default HTTPS port"
    )
  if graph_issued_url:
    allowed = is_graph_issued_host(hostname)
  else:
    allowed = hostname in _ALLOWED_DESTINATION_HOSTS
  if not allowed:
    raise GraphClientError(
      f"request destination host is not allowed: {hostname}"
    )
  return hostname
class _DestinationRedirectHandler(urllib.request.HTTPRedirectHandler):
  """Apply the destination policy before following each HTTP redirect."""
  def redirect_request(
    self,
    request_object: urllib.request.Request,
    file_pointer: Any,
    code: int,
    message: str,
    headers: Any,
    new_url: str,
  ) -> urllib.request.Request | None:
    graph_issued_url = bool(
      getattr(request_object, "_zwl_graph_issued_url", False)
    )
    destination_host = _guard_destination(
      new_url, graph_issued_url=graph_issued_url
    )
    redirected = super().redirect_request(
      request_object,
      file_pointer,
      code,
      message,
      headers,
      new_url,
    )
    if redirected is not None:
      setattr(redirected, "_zwl_graph_issued_url", graph_issued_url)
      _contacted_hosts.add(destination_host)
    return redirected
_DESTINATION_OPENER = urllib.request.build_opener(_DestinationRedirectHandler())
def _open_url(
  request_object: urllib.request.Request, timeout: float
) -> Any:
  return _DESTINATION_OPENER.open(request_object, timeout=timeout)
def load_credentials(
  env: Mapping[str, str] | None = None,
) -> tuple[str, str, str]:
  """Read Graph client credentials from environment variables only."""
  source = os.environ if env is None else env
  names = ("GRAPH_TENANT_ID", "GRAPH_CLIENT_ID", "GRAPH_CLIENT_SECRET")
  values: list[str] = []
  for name in names:
    value = source.get(name, "")
    if not value:
      raise CredentialError(f"missing required environment variable: {name}")
    values.append(value)
  return values[0], values[1], values[2]
def graph_url(path: str) -> str:
  """Turn a Graph API path into an absolute v1.0 URL."""
  if path.startswith("https://") or path.startswith("http://"):
    return path
  return f"{GRAPH_BASE_URL}/{path.lstrip('/')}"
def encode_path_segments(path_or_segments: str | Iterable[str]) -> str:
  """Percent-encode each item-path segment while preserving separators.
    Pass a string when slashes are separators. Pass an iterable when a segment
    itself contains a slash, which Graph requires to be encoded as ``%2F``.
    """
  if isinstance(path_or_segments, str):
    leading_slash = path_or_segments.startswith("/")
    trailing_slash = path_or_segments.endswith("/") and path_or_segments != "/"
    segments = (
      path_or_segments.strip("/").split("/")
      if path_or_segments.strip("/")
      else []
    )
    encoded = "/".join(urllib.parse.quote(segment, safe="") for segment in segments)
    if leading_slash:
      encoded = f"/{encoded}"
    if trailing_slash:
      encoded = f"{encoded}/"
    return encoded
  return "/".join(
    urllib.parse.quote(str(segment), safe="") for segment in path_or_segments
  )
def validate_remote_path(path: str, *, allow_empty: bool = False) -> str:
  """Decode once and reject unsafe drive-relative paths before transport."""
  if not isinstance(path, str):
    raise ConfigError("remote path must be a string")
  decoded = urllib.parse.unquote(path)
  for candidate in (path, decoded):
    if candidate == "" and allow_empty:
      continue
    if (
      not candidate or "\\" in candidate or ":" in candidate
      or any(unicodedata.category(char).startswith("C") for char in candidate)
      or any(segment in {"", ".", ".."} for segment in candidate.split("/"))
    ):
      raise ConfigError(f"invalid remote path {ascii(path)}: use nonempty relative segments without traversal or controls")
  return path
def rooted_remote_path(root: str, relative: str = "") -> str:
  """Resolve only a validated relative path beneath a validated root."""
  root = validate_remote_path(root, allow_empty=True)
  relative = validate_remote_path(relative, allow_empty=True)
  return "/".join(part for part in (root, relative) if part)
def collection_page(payload: Any, url: str) -> tuple[list[Any], str | None]:
  """Reject malformed pages and pagination that changes the selected route."""
  if not isinstance(payload, dict) or not isinstance(payload.get("value"), list):
    raise GraphClientError("malformed collection response: expected an object with a value list")
  if any(not isinstance(item, dict) for item in payload["value"]):
    raise GraphClientError("malformed collection response: every value item must be an object")
  link = payload.get("@odata.nextLink")
  if "@odata.nextLink" in payload and not isinstance(link, str):
    raise GraphClientError("malformed collection response: nextLink must be a string")
  if link:
    current, following = urllib.parse.urlsplit(url), urllib.parse.urlsplit(link)
    if (following.scheme, following.netloc, following.path) != (
      current.scheme, current.netloc, current.path
    ) or following.fragment:
      raise GraphClientError("collection nextLink leaves the selected root route")
  return payload["value"], link or None
def check_collection_url(url: str, seen: set[str]) -> None:
  """Bound a collection traversal before requesting its next page."""
  if url in seen:
    raise GraphClientError("collection pagination repeated a nextLink")
  if len(seen) >= MAX_COLLECTION_PAGES:
    raise GraphClientError(f"collection pagination exceeds {MAX_COLLECTION_PAGES} pages")
  seen.add(url)
class LocalOutputError(GraphClientError):
  """A local output could not be created or written."""
def open_new_text(path: str | Path):
  """Reserve private output before writes; never follow parent symlinks."""
  destination = Path(path)
  directory_fd = None
  try:
    flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
    directory_fd = os.open(destination.anchor or ".", flags)
    parents = destination.parts[1:-1] if destination.is_absolute() else destination.parts[:-1]
    for component in parents:
      child_fd = os.open(component, flags, dir_fd=directory_fd)
      os.close(directory_fd)
      directory_fd = child_fd
    output_fd = os.open(destination.name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory_fd)
    try:
      return os.fdopen(output_fd, "w", encoding="utf-8")
    except BaseException:
      os.close(output_fd)
      raise
  except (OSError, AttributeError) as exc:
    raise LocalOutputError(f"cannot create new output {ascii(str(path))}: existing files, symlink components, and missing parents are refused; use the physical path (for example /private/tmp) ({exc})") from exc
  finally:
    if directory_fd is not None:
      os.close(directory_fd)
def write_new_text(path: str | Path, text: str) -> None:
  with open_new_text(path) as output:
    output.write(text)
def _headers_dict(headers: Mapping[str, str] | Any | None) -> dict[str, str]:
  if headers is None:
    return {}
  try:
    return {str(key): str(value) for key, value in headers.items()}
  except AttributeError:
    return dict(headers)
def _decode_payload(raw: bytes, expect_json: bool) -> Any:
  if not expect_json:
    return raw
  if not raw:
    return None
  try:
    return json.loads(raw.decode("utf-8"))
  except (UnicodeDecodeError, json.JSONDecodeError):
    return None
def _safe_url(url: str) -> str:
  """Keep Graph routes useful while redacting credential-bearing URL parts."""
  parsed = urllib.parse.urlsplit(url)
  if parsed.hostname == "graph.microsoft.com":
    path = re.sub(
      r"/(drives|sites|items)/[^/]+",
      lambda match: f"/{match.group(1)}/<redacted>",
      parsed.path,
    )
    return urllib.parse.urlunsplit(
      (parsed.scheme, parsed.netloc, path, parsed.query, parsed.fragment)
    )
  if parsed.hostname == "login.microsoftonline.com":
    return f"{parsed.scheme}://{parsed.netloc}/<redacted>/oauth2/v2.0/token"
  return f"{parsed.scheme}://{parsed.netloc}/<redacted>"
def _load_fake_responses(path: str) -> list[dict[str, Any]]:
  try:
    with open(path, "r", encoding="utf-8") as handle:
      document = json.load(handle)
  except FileNotFoundError as exc:
    raise GraphClientError(f"fake response fixture not found: {path}") from exc
  except (OSError, json.JSONDecodeError) as exc:
    raise GraphClientError(
      f"cannot load fake response fixture {path}: {exc}"
    ) from exc
  responses = document.get("responses") if isinstance(document, dict) else None
  if not isinstance(responses, list):
    raise GraphClientError(
      f"fake response fixture {path} must contain a responses array"
    )
  if not all(isinstance(entry, dict) for entry in responses):
    raise GraphClientError(
      f"fake response fixture {path} contains a non-object response"
    )
  return responses
def _fake_entry_matches(
  entry: dict[str, Any],
  method: str,
  url: str,
  headers: Mapping[str, str],
  json_body: Any,
) -> bool:
  if str(entry.get("method", "")).upper() != method:
    return False
  if not url.startswith(str(entry.get("url_prefix", ""))):
    return False
  lowered = {key.lower(): value for key, value in headers.items()}
  required_headers = entry.get("match_headers") or {}
  if isinstance(required_headers, dict):
    for name, value in required_headers.items():
      if lowered.get(str(name).lower()) != str(value):
        return False
  absent_headers = entry.get("match_absent_headers") or []
  if isinstance(absent_headers, list):
    for name in absent_headers:
      if str(name).lower() in lowered:
        return False
  if "match_json" in entry and json_body != entry["match_json"]:
    return False
  return True
def _fake_response(
  method: str,
  url: str,
  expect_json: bool,
  headers: Mapping[str, str] | None = None,
  json_body: Any = None,
) -> tuple[int, Any, dict[str, str]]:
  global _fake_path, _fake_responses
  if not _fake_transport_enabled():
    raise GraphClientError(
      "fake transport called without both GRAPH_FAKE_RESPONSES and "
      f"{_FAKE_TRANSPORT_ACK_VAR}"
    )
  print(
    "WARNING: GRAPH FAKE TRANSPORT ENGAGED - no real Microsoft Graph "
    "network call is being made; this must never happen outside the "
    "test suite.",
    file=sys.stderr,
  )
  path = os.environ["GRAPH_FAKE_RESPONSES"]
  if _fake_path != path or _fake_responses is None:
    _fake_path = path
    _fake_responses = _load_fake_responses(path)
  selected_index = None
  selected = None
  for index, entry in enumerate(_fake_responses):
    if _fake_entry_matches(entry, method, url, headers or {}, json_body):
      selected_index = index
      selected = entry
      break
  if selected is None or selected_index is None:
    raise GraphClientError(f"no fake response matched {method} {_safe_url(url)}")
  if selected.get("once") is True:
    del _fake_responses[selected_index]
  try:
    status = int(selected["status"])
  except (KeyError, TypeError, ValueError) as exc:
    raise GraphClientError(
      f"fake response for {method} {_safe_url(url)} has invalid status"
    ) from exc
  headers = _headers_dict(selected.get("headers"))
  if expect_json:
    payload = selected.get("json")
  elif "body_base64" in selected:
    try:
      payload = base64.b64decode(str(selected["body_base64"]), validate=True)
    except (ValueError, base64.binascii.Error) as exc:
      raise GraphClientError(
        f"fake response for {method} {_safe_url(url)} has invalid body_base64"
      ) from exc
  elif "body" in selected:
    payload = str(selected["body"]).encode("utf-8")
  elif "json" in selected:
    payload = json.dumps(selected["json"]).encode("utf-8")
  else:
    payload = b""
  return status, payload, headers
def _network_response(
  method: str,
  url: str,
  *,
  headers: Mapping[str, str],
  data: bytes | None,
  timeout: float,
  expect_json: bool,
  graph_issued_url: bool,
) -> tuple[int, Any, dict[str, str]]:
  ordinary_headers = dict(headers)
  authorization = next(
    (
      ordinary_headers.pop(key)
      for key in list(ordinary_headers)
      if key.lower() == "authorization"
    ),
    None,
  )
  request_object = urllib.request.Request(
    url,
    data=data,
    headers=ordinary_headers,
    method=method,
  )
  if authorization is not None:
    request_object.add_unredirected_header("Authorization", authorization)
  setattr(request_object, "_zwl_graph_issued_url", graph_issued_url)
  try:
    with _open_url(request_object, timeout) as response:
      raw = response.read()
      status = int(getattr(response, "status", response.getcode()))
      return (
        status,
        _decode_payload(raw, expect_json),
        _headers_dict(response.headers),
      )
  except urllib.error.HTTPError as exc:
    try:
      raw = exc.read()
      return (
        int(exc.code),
        _decode_payload(raw, expect_json),
        _headers_dict(exc.headers),
      )
    finally:
      exc.close()
  except urllib.error.URLError as exc:
    reason = getattr(exc, "reason", exc)
    raise GraphClientError(
      f"network request failed for {method} {_safe_url(url)}: {reason}"
    ) from exc
def _retry_delay(_status: int, headers: Mapping[str, str], attempt: int) -> float:
  retry_after = next(
    (value for key, value in headers.items() if key.lower() == "retry-after"),
    None,
  )
  if retry_after is not None and str(retry_after).strip().isdigit():
    return float(int(str(retry_after).strip()))
  return float(2**attempt)
def _is_retriable(status: int) -> bool:
  return status == 429 or 500 <= status <= 599
def _write_is_retriable(status: int, headers: Mapping[str, str]) -> bool:
  return status == 429 or (
    status == 503 and any(key.lower() == "retry-after" for key in headers)
  )
def request(
  method: str,
  url: str,
  *,
  token: str | None = None,
  headers: Mapping[str, str] | None = None,
  data: bytes | None = None,
  json_body: Any = None,
  timeout: float = 60,
  max_retries: int = 4,
  throttle: float = 0,
  expect_json: bool = True,
  graph_issued_url: bool = False,
  write: bool = False,
) -> tuple[int, Any, dict[str, str]]:
  """Issue one HTTP request with bounded retry on 429 and 5xx statuses.
    ``max_retries`` counts retries after the initial request. Retry-After is
    honored only when it is an integer number of seconds. Other date forms use
    exponential backoff. A ``write`` is retried only on 429 or 503 with
    Retry-After; another 5xx, a transport error, or a 409/412 after an earlier
    attempt raises WriteOutcomeUnknown because the write may have committed.
    """
  normalized_method = method.upper()
  if max_retries < 0:
    raise ValueError("max_retries must be >= 0")
  if timeout <= 0:
    raise ValueError("timeout must be > 0")
  if throttle < 0:
    raise ValueError("throttle must be >= 0")
  if data is not None and json_body is not None:
    raise ValueError("data and json_body are mutually exclusive")
  destination_host = _guard_destination(
    url, graph_issued_url=graph_issued_url
  )
  _contacted_hosts.add(destination_host)
  request_headers = dict(headers or {})
  if token and not any(key.lower() == "authorization" for key in request_headers):
    request_headers["Authorization"] = f"Bearer {token}"
  request_data = data
  if json_body is not None:
    request_data = json.dumps(json_body).encode("utf-8")
    if not any(key.lower() == "content-type" for key in request_headers):
      request_headers["Content-Type"] = "application/json"
  if throttle:
    time.sleep(throttle)
  last_result: tuple[int, Any, dict[str, str]] | None = None
  for attempt in range(max_retries + 1):
    if _fake_transport_enabled():
      result = _fake_response(
        normalized_method,
        url,
        expect_json,
        headers=request_headers,
        json_body=json_body,
      )
    else:
      try:
        result = _network_response(
          normalized_method,
          url,
          headers=request_headers,
          data=request_data,
          timeout=timeout,
          expect_json=expect_json,
          graph_issued_url=graph_issued_url,
        )
      except (GraphClientError, OSError) as exc:
        if not write:
          raise
        raise WriteOutcomeUnknown(
          f"{exc}; {WRITE_OUTCOME_UNKNOWN}"
        ) from exc
    last_result = result
    status, _, response_headers = result
    if write:
      unknown = (attempt and status in {409, 412}) or (
        500 <= status <= 599
        and not _write_is_retriable(status, response_headers)
      )
      if unknown:
        raise WriteOutcomeUnknown(
          f"{normalized_method} {_safe_url(url)} returned HTTP "
          f"{status}; {WRITE_OUTCOME_UNKNOWN}",
          status=status,
        )
      retriable = _write_is_retriable(status, response_headers)
    else:
      retriable = _is_retriable(status)
    if not retriable or attempt == max_retries:
      return result
    time.sleep(_retry_delay(status, response_headers, attempt))
  if last_result is None:
    raise GraphClientError("request did not produce an HTTP result")
  return last_result
def acquire_token(
  *,
  timeout: float = 30,
  max_retries: int = 4,
  throttle: float = 0,
  env: Mapping[str, str] | None = None,
) -> str:
  """Acquire an application token with the OAuth client-credentials grant."""
  tenant_id, client_id, client_secret = load_credentials(env)
  token_url = (
    f"{LOGIN_BASE_URL}/{urllib.parse.quote(tenant_id, safe='')}/oauth2/v2.0/token"
  )
  form = urllib.parse.urlencode(
    {
      "client_id": client_id,
      "client_secret": client_secret,
      "scope": GRAPH_SCOPE,
      "grant_type": "client_credentials",
    }
  ).encode("utf-8")
  status, payload, headers = request(
    "POST",
    token_url,
    headers={"Content-Type": "application/x-www-form-urlencoded"},
    data=form,
    timeout=timeout,
    max_retries=max_retries,
    throttle=throttle,
  )
  if not 200 <= status < 300:
    raise GraphClientError(
      f"token acquisition failed with HTTP {status}",
      status=status,
      payload=payload,
      headers=headers,
    )
  access_token = payload.get("access_token") if isinstance(payload, dict) else None
  if not isinstance(access_token, str) or not access_token:
    raise GraphClientError(
      "token acquisition response did not contain access_token",
      status=status,
      payload=payload,
      headers=headers,
    )
  return access_token
def validate_upload_chunk(value: Any) -> int:
  if (type(value) is not int or value <= 0
      or value % UPLOAD_CHUNK_UNIT or value >= UPLOAD_REQUEST_LIMIT):
    raise ConfigError("upload chunk must be a positive integer multiple of 327680 bytes and less than 60 MiB")
  return value
def load_config(
  path_or_none: str | os.PathLike[str] | None,
  *,
  overrides: Mapping[str, Any] | None = None,
  env: Mapping[str, str] | None = None,
  require_target: bool = True,
) -> dict[str, Any]:
  """Load JSON configuration and apply non-None CLI overrides field by field."""
  source = os.environ if env is None else env
  selected_path = (
    str(path_or_none)
    if path_or_none
    else source.get("SHAREPOINT_DATAROOM_CONFIG", "")
  )
  loaded: dict[str, Any] = {}
  if selected_path:
    try:
      with open(selected_path, "r", encoding="utf-8") as handle:
        document = json.load(handle)
    except FileNotFoundError as exc:
      raise ConfigError(f"config file not found: {selected_path}") from exc
    except json.JSONDecodeError as exc:
      raise ConfigError(
        f"invalid JSON in config file {selected_path}: line {exc.lineno} column {exc.colno}"
      ) from exc
    except OSError as exc:
      raise ConfigError(
        f"cannot read config file {selected_path}: {exc}"
      ) from exc
    if not isinstance(document, dict):
      raise ConfigError(f"config file {selected_path} must contain a JSON object")
    loaded.update(document)
  merged = dict(CONFIG_DEFAULTS)
  merged.update(loaded)
  for key, value in (overrides or {}).items():
    if value is not None:
      merged[key] = value
  if require_target and not (merged.get("drive_id") or merged.get("site_id")):
    raise ConfigError(
      "neither drive_id nor site_id was provided by CLI flags or configuration"
    )
  root = merged.get("root_path", "")
  if isinstance(root, str):
    if root.startswith("//") or root.endswith("//"):
      validate_remote_path(root)
    if root.startswith("/"):
      root = root[1:]
    if root.endswith("/"):
      root = root[:-1]
  validate_upload_chunk(merged["upload_chunk_bytes"])
  merged["root_path"] = validate_remote_path(root, allow_empty=True)
  return merged
def resolve_drive_id(
  token: str,
  *,
  drive_id: str | None = None,
  site_id: str | None = None,
  timeout: float = 60,
  max_retries: int = 4,
  throttle: float = 0,
) -> str:
  """Resolve a site's default drive. Verify against Microsoft documentation."""
  if drive_id:
    return drive_id
  if not site_id:
    raise ConfigError(
      "neither drive_id nor site_id was provided by CLI flags or configuration"
    )
  url = graph_url(f"/sites/{urllib.parse.quote(site_id, safe='')}/drive")
  status, payload, headers = request(
    "GET",
    url,
    token=token,
    timeout=timeout,
    max_retries=max_retries,
    throttle=throttle,
  )
  if not 200 <= status < 300:
    raise GraphClientError(
      f"default drive resolution failed with HTTP {status}",
      status=status,
      payload=payload,
      headers=headers,
    )
  resolved = payload.get("id") if isinstance(payload, dict) else None
  if not isinstance(resolved, str) or not resolved:
    raise GraphClientError(
      "default drive response did not contain an id",
      status=status,
      payload=payload,
      headers=headers,
    )
  return resolved
def mime_type_for(path: str | os.PathLike[str]) -> str:
  """Return a conservative content type for an upload path."""
  suffix = Path(path).suffix.lower()
  if suffix in MIME_TYPE_OVERRIDES:
    return MIME_TYPE_OVERRIDES[suffix]
  guessed, _ = mimetypes.guess_type(str(path))
  return guessed or "application/octet-stream"
def uses_chunked_upload(size_bytes: int, small_threshold: int) -> bool:
  """Return true only when a file is larger than the direct-upload limit."""
  if size_bytes < 0:
    raise ValueError("size_bytes must be >= 0")
  if small_threshold < 0:
    raise ValueError("small_threshold must be >= 0")
  return size_bytes > small_threshold
def upload_method(size_bytes: int, small_threshold: int) -> str:
  """Return the transport label used by audit records."""
  return "chunked" if uses_chunked_upload(size_bytes, small_threshold) else "direct"
def display_text(value: Any) -> str:
  """Render a value as one unambiguous line of human-readable text.
    Local paths, remote names, eTags, ids and error text can contain line
    breaks or other control characters that would forge extra output lines.
    A backslash is doubled, and every Unicode category C character (controls,
    format characters, surrogates, private use, unassigned) plus the line and
    paragraph separators U+2028 and U+2029 is written as a visible \\uXXXX or
    \\UXXXXXXXX escape. JSON output is not passed through this function.
    """
  rendered: list[str] = []
  for character in str(value):
    codepoint = ord(character)
    if character == "\\":
      rendered.append("\\\\")
    elif (
      unicodedata.category(character).startswith("C")
      or codepoint in (0x2028, 0x2029)
    ):
      if codepoint <= 0xFFFF:
        rendered.append(f"\\u{codepoint:04x}")
      else:
        rendered.append(f"\\U{codepoint:08x}")
    else:
      rendered.append(character)
  return "".join(rendered)

License: MIT (read it).

Back to the listing