SharePoint / OneDrive × Python── Reading drives, lists and column values with Graph, and tracking only changes

"""Resolve a SharePoint site id from its URL (hostname + server-relative path)."""
import asyncio
import os
from urllib.parse import urlparse

from azure.identity import DeviceCodeCredential
from msgraph import GraphServiceClient
from msgraph.generated.sites.item.site_item_request_builder import SiteItemRequestBuilder

SCOPES = ["Sites.Read.All"]
SITE_URL = os.environ.get("SITE_URL", "https://example.sharepoint.com/sites/docs")


def site_path_url(site_url):
    # by_site_id() percent-encodes ":" and "/", so build the documented URL shape
    u = urlparse(site_url)
    return "https://graph.microsoft.com/v1.0/sites/%s:%s" % (u.hostname, u.path.rstrip("/"))


async def main():
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    builder = SiteItemRequestBuilder(client.request_adapter, site_path_url(SITE_URL))
    site = await builder.get()
    print("id:", site.id)
    print("name:", site.display_name)
    print("webUrl:", site.web_url)


if __name__ == "__main__":
    asyncio.run(main())

"""List a site's document libraries, then every item in one folder of a library."""
import asyncio
import csv
import os
from urllib.parse import quote

from azure.identity import DeviceCodeCredential
from msgraph import GraphServiceClient

SCOPES = ["Files.Read.All"]
LIBRARY = os.environ.get("LIBRARY_NAME", "Documents")
FOLDER_PATH = os.environ.get("FOLDER_PATH", "Reports")
SELECT = "name,size,lastModifiedDateTime,webUrl,folder,file"


def to_row(item):
    kind = "folder" if item.folder else "file"
    return [kind, item.name, item.size or 0, item.last_modified_date_time, item.web_url]


async def main():
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    drives = await client.sites.by_site_id(os.environ["SITE_ID"]).drives.get()
    drive = next((d for d in drives.value or [] if d.name == LIBRARY), None)
    if drive is None:
        print("library not found:", LIBRARY, [d.name for d in drives.value or []])
        return
    # documented path form; the SDK would encode ":" inside an item id
    url = "https://graph.microsoft.com/v1.0/drives/%s/root:/%s:/children?$select=%s" % (
        drive.id, quote(FOLDER_PATH), SELECT)
    builder = client.drives.by_drive_id(drive.id).items.by_drive_item_id("root").children
    page = await builder.with_url(url).get()
    rows = []
    while page:
        rows += [to_row(i) for i in page.value or []]
        if not page.odata_next_link:
            break
        page = await builder.with_url(page.odata_next_link).get()
    with open("out_children.csv", "w", encoding="utf-8-sig", newline="") as f:
        w = csv.writer(f)
        w.writerow(["kind", "name", "size", "modified", "webUrl"])
        w.writerows(rows)
    print("items:", len(rows))


if __name__ == "__main__":
    asyncio.run(main())

"""Download one file from a drive to a new local file (never overwrites)."""
import asyncio
import os
from pathlib import Path

from azure.identity import DeviceCodeCredential
from kiota_abstractions.base_request_configuration import RequestConfiguration
from msgraph import GraphServiceClient
from msgraph.generated.drives.item.items.item.drive_item_item_request_builder import (
    DriveItemItemRequestBuilder,
)

SCOPES = ["Files.Read.All"]


def target_path(name, folder="."):
    # keep only the file name and add an out_ prefix; refuse to overwrite
    path = Path(folder) / ("out_" + Path(name).name)
    if path.exists():
        raise FileExistsError(path)
    return path


async def main():
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    item_rb = client.drives.by_drive_id(os.environ["DRIVE_ID"]).items.by_drive_item_id(os.environ["ITEM_ID"])
    query = DriveItemItemRequestBuilder.DriveItemItemRequestBuilderGetQueryParameters(
        select=["name", "size", "file"])
    item = await item_rb.get(request_configuration=RequestConfiguration(query_parameters=query))
    if item.file is None:
        print("not a file:", item.name)
        return
    data = await item_rb.content.get()
    path = target_path(item.name)
    path.write_bytes(data)
    print("saved", path, len(data), "bytes")


if __name__ == "__main__":
    asyncio.run(main())

"""Read list items with selected column values and write them to CSV."""
import asyncio
import csv
import os

from azure.identity import DeviceCodeCredential
from kiota_abstractions.base_request_configuration import RequestConfiguration
from msgraph import GraphServiceClient
from msgraph.generated.sites.item.lists.item.items.items_request_builder import ItemsRequestBuilder

SCOPES = ["Sites.Read.All"]
COLUMNS = ["FileLeafRef", "DocId", "Status", "ReviewDue", "Owner"]


def flatten(item_id, web_url, fields):
    row = {"id": item_id, "webUrl": web_url}
    for c in COLUMNS:
        row[c] = fields.get(c)
    return row


async def main():
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    items_rb = client.sites.by_site_id(os.environ["SITE_ID"]).lists.by_list_id(os.environ["LIST_ID"]).items
    query = ItemsRequestBuilder.ItemsRequestBuilderGetQueryParameters(
        expand=["fields(select=%s)" % ",".join(COLUMNS)])
    page = await items_rb.get(request_configuration=RequestConfiguration(query_parameters=query))
    rows = []
    while page:
        for it in page.value or []:
            fields = it.fields.additional_data if it.fields else {}
            rows.append(flatten(it.id, it.web_url, fields))
        if not page.odata_next_link:
            break
        page = await items_rb.with_url(page.odata_next_link).get()
    with open("out_list_items.csv", "w", encoding="utf-8-sig", newline="") as f:
        w = csv.DictWriter(f, fieldnames=["id", "webUrl"] + COLUMNS)
        w.writeheader()
        w.writerows(rows)
    print("rows:", len(rows))


if __name__ == "__main__":
    asyncio.run(main())

"""Track changes in a drive: save the deltaLink and resume from it on the next run."""
import asyncio
import json
import os
from pathlib import Path

from azure.identity import DeviceCodeCredential
from kiota_abstractions.api_error import APIError
from msgraph import GraphServiceClient

SCOPES = ["Files.Read.All"]
STATE = Path(os.environ.get("DELTA_STATE", "out_delta_state.json"))


def first_link(drive_id):
    return "https://graph.microsoft.com/v1.0/drives/%s/root/delta" % drive_id


def load_link(drive_id):
    if STATE.exists():
        return json.loads(STATE.read_text(encoding="utf-8"))["deltaLink"]
    return first_link(drive_id)


def summarize(items):
    changed = [i["name"] for i in items if not i["deleted"]]
    removed = [i["id"] for i in items if i["deleted"]]
    return changed, removed


async def read_all(builder, link):
    items, page = [], await builder.with_url(link).get()
    while True:
        items += [{"id": i.id, "name": i.name, "deleted": i.deleted is not None} for i in page.value or []]
        if not page.odata_next_link:
            return items, page.odata_delta_link
        page = await builder.with_url(page.odata_next_link).get()


async def main():
    drive_id = os.environ["DRIVE_ID"]
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    builder = client.drives.by_drive_id(drive_id).items.by_drive_item_id("root").delta
    try:
        items, delta_link = await read_all(builder, load_link(drive_id))
    except APIError as e:
        if e.response_status_code != 410:
            raise
        # the saved token can no longer be used: enumerate everything again
        items, delta_link = await read_all(builder, first_link(drive_id))
    changed, removed = summarize(items)
    print("changed:", len(changed), "deleted:", len(removed))
    for name in changed[:10]:
        print(" ", name)
    STATE.write_text(json.dumps({"deltaLink": delta_link}), encoding="utf-8")


if __name__ == "__main__":
    asyncio.run(main())

"""App-only read of a site that was assigned to this app through Sites.Selected."""
import asyncio
import os

from azure.identity import CertificateCredential, ManagedIdentityCredential
from msgraph import GraphServiceClient

# app-only tokens use .default; the app registration holds Sites.Selected
SCOPES = ["https://graph.microsoft.com/.default"]


def make_credential():
    cert = os.environ.get("CERT_PATH", "")
    if cert and os.path.exists(cert):
        return CertificateCredential(
            tenant_id=os.environ["TENANT_ID"], client_id=os.environ["CLIENT_ID"], certificate_path=cert)
    # on Azure (Functions, VM) use a managed identity instead of a certificate file
    return ManagedIdentityCredential(client_id=os.environ.get("AZURE_CLIENT_ID"))


def describe(site_name, lists):
    lines = ["site: %s" % site_name]
    lines += ["  list: %s  %s" % (name, url) for name, url in lists]
    return "\n".join(lines)


async def main():
    client = GraphServiceClient(credentials=make_credential(), scopes=SCOPES)
    site_rb = client.sites.by_site_id(os.environ["SITE_ID"])
    site = await site_rb.get()
    lists = await site_rb.lists.get()
    print(describe(site.display_name, [(x.display_name, x.web_url) for x in lists.value or []]))


if __name__ == "__main__":
    asyncio.run(main())

"""List anyone/organization sharing links on the files of one folder (read only)."""
import asyncio
import csv
import os
from urllib.parse import quote

from azure.identity import DeviceCodeCredential
from msgraph import GraphServiceClient

SCOPES = ["Files.Read.All"]
FOLDER_PATH = os.environ.get("FOLDER_PATH", "Reports")
BROAD = {"anonymous", "organization"}


def broad_links(file_name, perms):
    out = []
    for p in perms:
        link = p.get("link") or {}
        if link.get("scope") in BROAD:
            out.append([file_name, link["scope"], link.get("type"), ",".join(p.get("roles") or []),
                        p.get("expirationDateTime") or ""])
    return out


def as_dict(p):
    link = {"scope": p.link.scope, "type": p.link.type} if p.link else None
    exp = p.expiration_date_time.isoformat() if p.expiration_date_time else ""
    return {"roles": p.roles, "link": link, "expirationDateTime": exp}


async def main():
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    drive_rb = client.drives.by_drive_id(os.environ["DRIVE_ID"])
    url = "https://graph.microsoft.com/v1.0/drives/%s/root:/%s:/children?$select=id,name,file" % (
        os.environ["DRIVE_ID"], quote(FOLDER_PATH))
    children_rb = drive_rb.items.by_drive_item_id("root").children
    page, files = await children_rb.with_url(url).get(), []
    while page:
        files += [i for i in page.value or [] if i.file]
        if not page.odata_next_link:
            break
        page = await children_rb.with_url(page.odata_next_link).get()
    rows = []
    for f in files:
        perms = await drive_rb.items.by_drive_item_id(f.id).permissions.get()
        rows += broad_links(f.name, [as_dict(p) for p in perms.value or []])
    with open("out_broad_links.csv", "w", encoding="utf-8-sig", newline="") as fh:
        w = csv.writer(fh)
        w.writerow(["file", "scope", "type", "roles", "expires"])
        w.writerows(rows)
    print("files:", len(files), "broad links:", len(rows))


if __name__ == "__main__":
    asyncio.run(main())

"""Audit a library: empty required columns and past-due dates (report only)."""
import asyncio
import csv
import datetime
import os

from azure.identity import DeviceCodeCredential
from kiota_abstractions.base_request_configuration import RequestConfiguration
from msgraph import GraphServiceClient
from msgraph.generated.sites.item.lists.item.items.items_request_builder import ItemsRequestBuilder

SCOPES = ["Sites.Read.All"]
DATE_COL = os.environ.get("DATE_COLUMN", "ReviewDue")


def audit(items, required, date_col, today):
    rows = []
    for it in items:
        f = it.get("fields") or {}
        name = f.get("FileLeafRef", it.get("id"))
        for col in required:
            if f.get(col) in (None, ""):
                rows.append([it.get("id"), name, "missing:" + col, it.get("webUrl")])
        due = f.get(date_col)
        if due and str(due)[:10] < today:
            rows.append([it.get("id"), name, "overdue", it.get("webUrl")])
    return rows


async def fetch(client, site_id, list_id):
    list_rb = client.sites.by_site_id(site_id).lists.by_list_id(list_id)
    cols = await list_rb.columns.get()
    required = [c.name for c in cols.value or [] if c.required and not c.hidden and not c.read_only]
    select = ",".join(["FileLeafRef", DATE_COL] + required)
    query = ItemsRequestBuilder.ItemsRequestBuilderGetQueryParameters(expand=["fields(select=%s)" % select])
    page = await list_rb.items.get(request_configuration=RequestConfiguration(query_parameters=query))
    items = []
    while page:
        items += [{"id": i.id, "webUrl": i.web_url, "fields": i.fields.additional_data if i.fields else {}}
                  for i in page.value or []]
        if not page.odata_next_link:
            break
        page = await list_rb.items.with_url(page.odata_next_link).get()
    return items, required


async def main():
    cred = DeviceCodeCredential(client_id=os.environ["CLIENT_ID"], tenant_id=os.environ["TENANT_ID"])
    client = GraphServiceClient(credentials=cred, scopes=SCOPES)
    items, required = await fetch(client, os.environ["SITE_ID"], os.environ["LIST_ID"])
    rows = audit(items, required, DATE_COL, datetime.date.today().isoformat())
    with open("out_column_audit.csv", "w", encoding="utf-8-sig", newline="") as f:
        w = csv.writer(f)
        w.writerow(["id", "file", "problem", "webUrl"])
        w.writerows(rows)
    print("required:", required, "problems:", len(rows))


if __name__ == "__main__":
    asyncio.run(main())