graphbridge 0.0.5

GraphBridge is a lightweight Microsoft Graph client that uses app-only (Azure AD) authentication and streamlines SharePoint site/list operations: metadata retrieval, feature-based queries, CRUD, upsert, and field key encoding/decoding.

GraphBridge

A tiny Python helper to authenticate against Microsoft Graph (app-only) and work with SharePoint Online (sites & lists): read, create, update, delete, upsert, and batch.

Works with Python 3.10+ · Microsoft Graph v1.0 · App-only auth via client credentials (Entra ID / Azure AD)

---

Table of Contents

• Installation
• Microsoft 365 prerequisites
• Key concepts
• Quick start
• Usage examples
  • Read list rows
  • Create, update, delete
  • Upsert with upload()
  • Batch: create_many() and delete_many()
  • Filter with get_items_by_features()
  • Handle fields with spaces/symbols
• API reference
• Errors & best practices
• Security
• Compatibility & notes
• Full example

---

Installation

pip install graphbridge azure-identity requests

If your package on PyPI uses a different name, replace graphbridge with the actual name. Runtime dependencies: azure-identity and requests.

---

Microsoft 365 prerequisites

1. An App registration in Entra ID (Azure AD) with Client ID, Tenant ID, and Client Secret.

2. Application permissions for Microsoft Graph (SharePoint scope):

   • Read-only: Sites.Read.All
   • Read/Write: Sites.ReadWrite.All

3. Admin consent granted by a tenant admin.

4. The app must have access to the target SharePoint site and list.

---

Key concepts

• GbAuth Builds app-only authentication (client credentials) and provides the Bearer token and headers.

• GbSite Resolves a SharePoint site from hostname (e.g., contoso.sharepoint.com) and site_path (e.g., /sites/Marketing) to obtain the site_id via Graph.

• GbList Operates on a SharePoint list: read items/rows, create, update, delete, upsert, and batch via Graph $batch.

---

Quick start

from graphbridge import GbAuth, GbSite, GbList
import os

# 1) App-only authentication
auth = GbAuth(
    tenant_id=os.environ["AZURE_TENANT_ID"],
    client_id=os.environ["AZURE_CLIENT_ID"],
    client_secret=os.environ["AZURE_CLIENT_SECRET"],
)

# 2) Resolve the SharePoint site
site = GbSite(
    hostname="contoso.sharepoint.com",
    site_path="/sites/Marketing",
    gb_auth=auth,
)
print("Site ID:", site.site_id)

# 3) Bind the list
tasks = GbList(
    list_name="Tasks",   # display name of your list
    gb_site=site
)

# 4) Read the first rows
rows = tasks.list_rows
print("Example row:", rows[0] if rows else "No items")

---

Usage examples

Read list rows

rows = tasks.list_rows        # list of dicts under 'fields'
ids = tasks.list_ids          # list item IDs
columns = tasks.list_fields   # column names (keys from first 'fields')

for r in rows[:5]:
    print(r.get("Title"), r.get("Status"))

list_rows relies on list_items_all, which automatically paginates and returns all items in the list.

---

Create, update, delete

Create one or more items

# Single
res_create = tasks.create({"Title": "New task", "Status": "Open"})
print(res_create)

# Multiple
res_create_many = tasks.create([
    {"Title": "Task 1", "Status": "Open"},
    {"Title": "Task 2", "Status": "InProgress"},
])
print(res_create_many)

Update by ID (PATCH fields)

res_update = tasks.update(
    ids="25",
    rows={"Status": "Done", "PercentComplete": 1.0}
)
print(res_update)

Delete by ID

res_delete = tasks.delete(ids=["25", "26", "27"])
print(res_delete)

Every method returns a result object with successes and failures so you can inspect what happened.

---

Upsert with upload()

upload(ids, rows, force=False, delete=False) synchronizes the list with your local source:

• If an id already exists:

  • force=True ➜ delete & recreate the item (replace).
  • force=False ➜ patch update the item.

• If an id does not exist ➜ create a new item.

• delete=True ➜ remove items not present in ids (cleanup).

# Bring the list in sync with 3 source rows:
ids = ["101", "102", "103"]
rows = [
    {"Title": "A", "Status": "Open"},
    {"Title": "B", "Status": "InProgress"},
    {"Title": "C", "Status": "Done"},
]

res_upload = tasks.upload(ids=ids, rows=rows, force=False, delete=True)
print(res_upload)

Simplified return shape:

{
  "delete_results": {"successes": [], "failures": []},
  "force_results": {
    "replaced": {"successes": [], "failures": []},
    "updated":  {"successes": [{"id":"101","row":{"Title":"A"}}], "failures": []},
    "created":  {"successes": [{"id":"103","new_id":"345"}], "failures": []}
  }
}

---

Batch: create_many() and delete_many()

These use the Graph $batch endpoint (default 20 sub-requests per batch).

# CREATE in batch
bulk_rows = [{"Title": f"Bulk {i}"} for i in range(1, 51)]
bulk_create = tasks.create_many(bulk_rows, batch_size=20)
print(bulk_create)

# DELETE in batch
bulk_delete = tasks.delete_many(ids=["301", "302", "303"], batch_size=20, if_match="*")
print(bulk_delete)

if_match="*" disables concurrency checks for batch deletes (use carefully).

---

Filter with get_items_by_features()

Returns raw list items (not only fields) that match at least one criteria dict (OR across dicts; AND inside each dict).

Important: matching occurs on the top-level keys of each item. To filter by list fields, include the "fields" key explicitly.

features = [
    {"fields": {"Status": "Open"}},  # match on fields.Status
    {"id": "123"}                    # match on top-level id
]

matched_items = tasks.get_items_by_features(features)
print(len(matched_items))

---

Handle fields with spaces/symbols

SharePoint often encodes internal column names (e.g., spaces) as _x0020_. Use the helpers to map keys both ways:

row = {"Project Name": "Apollo", "Start-Date": "2025-08-24"}

encoded = tasks.encode_row(row)   # {'Project_x0020_Name': 'Apollo', 'Start_x002d_Date': '2025-08-24'}
decoded = tasks.decode_row(encoded)  # back to human-friendly keys

This is handy when an API/payload requires internal field names.

---

API reference

Utility

• deduplicate_dicts(dict_list: list[dict]) -> list[dict] Removes duplicates (by sorted JSON) from a list of dicts.

---

class GbAuth

GbAuth(tenant_id: str, client_id: str, client_secret: str)

Properties:

• credential: ClientSecretCredential – lazily cached azure.identity.ClientSecretCredential.
• token: str – Graph access token for https://graph.microsoft.com/.default.
• headers: dict – {"Authorization": f"Bearer {token}"}.

Validation: raises TypeError/ValueError for empty/non-string inputs; raises RuntimeError if token acquisition fails.

---

class GbSite(GbAuth)

GbSite(hostname: str, site_path: str, gb_auth: GbAuth | None = None, **auth_kwargs)

• Pass either an existing GbAuth via gb_auth or auth keywords (tenant_id, client_id, client_secret).

Properties:

• hostname: str – e.g., contoso.sharepoint.com
• site_path: str – e.g., /sites/Marketing (or /teams/...)
• site_url: str – Graph URL for the site.
• site_data: dict – site metadata (cached).
• site_id: str – resolved from site_data.

Errors: raises RuntimeError if the site GET is not 200.

---

class GbList(GbSite)

GbList(list_name: str, gb_site: GbSite | None = None, **site_and_auth_kwargs)

• Pass either a GbSite or site+auth keywords (hostname, site_path, tenant_id, …).

Core properties:

• list_url: str, list_data: dict, list_id: str
• list_items_all -> list[dict] All list items with automatic pagination. Internally requests pages of 200 items; page size is not configurable via the property.
• list_items -> list[dict] First page of items (handy for quick checks).
• list_rows -> list[dict] Only the fields section of each item (derived from list_items_all).
• list_ids -> list[str]
• list_fields -> list[str] (keys from the first fields if present)
• encode_row(row: dict) -> dict, decode_row(row: dict) -> dict Bidirectional mapping between friendly keys and encoded _x00.._ internal names.

CRUD methods:

• create(rows: dict | list[dict] | tuple[dict] | set[dict]) -> dict
• update(ids: str | int | list[str] | tuple[str] | set[str], rows: dict | list[dict] | tuple[dict]) -> dict
• delete(ids: str | list[str] | tuple[str] | set[str]) -> dict

Upsert & batch:

• upload(ids, rows, force=False, delete=False) -> dict
• create_many(rows: list[dict], batch_size: int = 20) -> dict
• delete_many(ids, batch_size: int = 20, if_match: str | None = None) -> dict

Criteria-based retrieval:

• get_items_by_features(features: list[dict]) -> list[dict] OR across dicts, AND within each dict. Note: comparisons are on top-level item keys; use {"fields": {...}} for list fields.

---

Errors & best practices

• Methods that call Graph return:

  • successes: items with details (id, item, updated_row, …)
  • failures: errors with status/error payload

• Wrap critical calls with try/except:

try:
    out = tasks.create({"Title": "Check errors"})
    if out["failures"]:
        print("Failures:", out["failures"])
except Exception as e:
    print("Fatal error:", e)

• Permissions: 403 Forbidden usually means missing application permission (Sites.ReadWrite.All) or missing admin consent.
• Concurrency: batch deletes support if_match; standard updates use PATCH on fields.
• Pagination: prefer list_rows/list_items_all to fetch complete sets; list_items is a single page.
• Rate limits: for large workloads, prefer batch methods and keep batch_size reasonable (typically ≤ 20).

---

Security

• Never hardcode the Client Secret. Use environment variables or a secret vault.
• GbAuth obtains tokens at runtime via azure-identity.
• Never commit credentials or tokens to source control.

---

Compatibility & notes

• Python 3.10+ (uses modern union types like str | int).
• list_items_all is exposed as a property; it internally requests pages of 200 items until completion.
• Column names: Graph’s fields often use internal names; if your columns have spaces/symbols, use encode_row/decode_row to simplify payload handling.

---

Full example

from graphbridge import GbAuth, GbSite, GbList

auth = GbAuth(
    tenant_id="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
    client_id="yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy",
    client_secret="********"
)

site = GbSite(
    hostname="contoso.sharepoint.com",
    site_path="/sites/Finance",
    gb_auth=auth
)

invoices = GbList(list_name="Invoices", gb_site=site)

# 1) Read
for r in invoices.list_rows[:3]:
    print("Invoice:", r.get("Title"), r.get("Amount"))

# 2) Create
new_items = invoices.create([
    {"Title": "I-2025-001", "Amount": 1000},
    {"Title": "I-2025-002", "Amount": 2500},
])
print("Create:", new_items["successes"])

# 3) Update
upd = invoices.update(ids="42", rows={"Amount": 3000})
print("Update:", upd)

# 4) Upsert & cleanup
ids = ["1001", "1002"]
rows = [{"Title": "A"}, {"Title": "B"}]
print(invoices.upload(ids=ids, rows=rows, force=False, delete=True))

# 5) Batch delete
print(invoices.delete_many(ids=["1001", "1002"], if_match="*"))