Add migration examples: export/import users/groups/sites to CSV

Vadim · Vadim · commit 9d6879b9456a · 2026-06-09T11:05:14.000Z
diff --git a/examples/entraid/migration/export_groups_csv.py b/examples/entraid/migration/export_groups_csv.py
@@ -0,0 +1,80 @@
+"""
+Export groups with members to CSV — group structure for migration
+mapping and planning.
+
+Exports all groups with their membership list, ownership, group
+type, visibility, and creation date. Useful for:
+  - Mapping group structure during tenant-to-tenant migration
+  - Membership audit and cleanup
+  - Security group review
+
+Requires delegated permission ``Group.Read.All``,
+``Directory.Read.All``.
+
+https://learn.microsoft.com/en-us/graph/api/group-list
+"""
+
+import csv
+import sys
+
+from office365.graph_client import GraphClient
+from tests import test_client_id, test_client_secret, test_tenant
+
+OUTPUT_FILE = "/tmp/groups_export.csv"
+
+
+def main():
+    client = GraphClient(tenant=test_tenant).with_client_secret(test_client_id, test_client_secret)
+
+    print("Fetching groups...")
+    groups = client.groups.get_all().execute_query()
+    print(f"  Found {len(groups)} groups\n")
+
+    with open(OUTPUT_FILE, "w", newline="") as f:
+        writer = csv.writer(f)
+        writer.writerow([
+            "groupId", "displayName", "mail", "groupType", "visibility",
+            "description", "createdDateTime", "memberCount",
+            "members (UPN)", "owners (UPN)",
+        ])
+
+        for g in groups:
+            gid = g.id or "?"
+            display = g.display_name or "(unnamed)"
+            mail = g.mail or ""
+            gtype = "Microsoft 365" if g.mail_enabled and not g.security_enabled else \
+                    "Security" if g.security_enabled else "Distribution"
+            visibility = g.visibility or "?"
+            desc = (g.description or "")[:80]
+            created = g.created_date_time.strftime("%Y-%m-%d") if g.created_date_time else ""
+
+            # Load members and owners
+            member_upns = []
+            owner_upns = []
+
+            try:
+                members = g.members.get().execute_query()
+                member_upns = [m.user_principal_name for m in members if hasattr(m, "user_principal_name")][:20]
+            except Exception:
+                pass
+
+            try:
+                owners = g.owners.get().execute_query()
+                owner_upns = [o.user_principal_name for o in owners if hasattr(o, "user_principal_name")]
+            except Exception:
+                pass
+
+            writer.writerow([
+                gid, display, mail, gtype, visibility,
+                desc, created, len(member_upns),
+                ";".join(member_upns) if member_upns else "",
+                ";".join(owner_upns) if owner_upns else "",
+            ])
+
+            print(f"  {display:40s}  type={gtype:15s}  members={len(member_upns):>4}  owners={len(owner_upns)}")
+
+    print(f"\n✓ Exported {len(groups)} groups to {OUTPUT_FILE}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/entraid/migration/export_sites_csv.py b/examples/entraid/migration/export_sites_csv.py
@@ -0,0 +1,80 @@
+"""
+Export SharePoint sites to CSV — site inventory for migration
+planning and tenant management.
+
+Exports all SharePoint Online sites with URL, title, template,
+owner, storage usage/quota, compatibility level, hub association,
+and last content modified date.
+
+Essential for:
+  - SharePoint migration planning (source inventory)
+  - Tenant consolidation or split
+  - Storage governance and legacy site identification
+
+Requires delegated permission ``Sites.Read.All``.
+
+https://learn.microsoft.com/en-us/sharepoint/dev/apis/rest-api/tenant/GetSitePropertiesFromSharePoint
+"""
+
+import csv
+import sys
+
+from office365.sharepoint.client_context import ClientContext
+from office365.sharepoint.tenant.administration.tenant import Tenant
+from tests import test_admin_site_url, test_client_id, test_client_secret, test_tenant
+
+OUTPUT_FILE = "/tmp/sharepoint_sites_export.csv"
+_KB = 1024
+
+
+def format_mb(mb_value: int) -> str:
+    if mb_value >= _KB:
+        return f"{mb_value / _KB:.1f} GB"
+    return f"{mb_value} MB"
+
+
+def main():
+    ctx = ClientContext(test_admin_site_url).with_client_secret(test_tenant, test_client_id, test_client_secret)
+    tenant = Tenant(ctx)
+
+    print("Fetching SharePoint sites...")
+    sites = tenant.get_site_properties_from_sharepoint().execute_query()
+    print(f"  Found {len(sites)} sites\n")
+
+    with open(OUTPUT_FILE, "w", newline="") as f:
+        writer = csv.writer(f)
+        writer.writerow([
+            "url", "title", "template", "owner", "storageQuotaMB",
+            "storageUsageMB", "storageUsagePct", "compatibilityLevel",
+            "sharingCapability", "lockState", "lastContentModifiedDate",
+            "hubSite", "createdDate", "websCount",
+        ])
+
+        for site in sites:
+            url = getattr(site, "url", "?")
+            title = getattr(site, "title", "(untitled)")
+            template = getattr(site, "template", "?")
+            owner = getattr(site, "owner_login_name", "?") or getattr(site, "owner", "?")
+            quota = getattr(site, "storage_quota", 0) or 0
+            usage = getattr(site, "storage_usage_current", 0) or 0
+            pct = round((usage / quota) * 100, 1) if quota > 0 else 0
+            compat = getattr(site, "compatibility_level", "?")
+            sharing = getattr(site, "sharing_capability", "?")
+            lock = getattr(site, "lock_state", "")
+            last_mod = getattr(site, "last_content_modified_date", "")
+            if hasattr(last_mod, "strftime"):
+                last_mod = last_mod.strftime("%Y-%m-%d")
+            created = getattr(site, "created_time", "")
+            if hasattr(created, "strftime"):
+                created = created.strftime("%Y-%m-%d")
+            is_hub = getattr(site, "is_hub_site", False)
+            webs = getattr(site, "webs_count", "?")
+
+            writer.writerow([url, title, template, owner, quota, usage, pct, compat, sharing, lock, last_mod, is_hub, created, webs])
+            print(f"  {title[:40]:40s}  {format_mb(usage):>10s} / {format_mb(quota):>10s}  ({pct}%)")
+
+    print(f"\n✓ Exported {len(sites)} sites to {OUTPUT_FILE}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/entraid/migration/export_users_csv.py b/examples/entraid/migration/export_users_csv.py
@@ -0,0 +1,108 @@
+"""
+Export users to CSV — full user inventory for migration planning.
+
+Exports key user properties to a CSV file for integration with
+migration tools, HR systems, or audit reporting.
+
+Fields: UPN, display name, given name, surname, department, job
+title, office location, city, country, mobile/phone, enabled
+status, mail, MFA registered, creation date, last sign-in date,
+assigned license SKUs.
+
+Requires delegated permission ``User.Read.All``,
+``Organization.Read.All``, ``AuditLog.Read.All`` (for last sign-in),
+``UserAuthenticationMethod.Read.All`` (for MFA status).
+
+https://learn.microsoft.com/en-us/graph/api/user-list
+"""
+
+import csv
+import sys
+
+from office365.graph_client import GraphClient
+from tests import test_client_id, test_client_secret, test_tenant
+
+OUTPUT_FILE = "/tmp/users_export.csv"
+FIELDS = [
+    "userPrincipalName", "displayName", "givenName", "surname",
+    "department", "jobTitle", "officeLocation", "city", "country",
+    "mobilePhone", "businessPhones", "mail", "accountEnabled",
+    "userType", "createdDateTime", "signInActivity",
+    "assignedLicenses",
+]
+
+
+def get_license_skus(client: GraphClient) -> dict:
+    """Map SKU IDs to human-readable names."""
+    skus = client.subscribed_skus.get().execute_query()
+    return {s.sku_id: s.sku_part_number for s in skus if hasattr(s, "sku_id") and hasattr(s, "sku_part_number")}
+
+
+def get_mfa_status(client: GraphClient) -> dict[str, dict]:
+    """Get MFA and passwordless registration status per user."""
+    try:
+        details = client.reports.authentication_methods.user_registration_details.get().execute_query()
+        return {d.user_principal_name.upper(): d for d in details}
+    except Exception:
+        return {}
+
+
+def main():
+    client = GraphClient(tenant=test_tenant).with_client_secret(test_client_id, test_client_secret)
+
+    print("Fetching users...")
+    users = client.users.get_all().execute_query()
+    print(f"  Found {len(users)} users")
+
+    sku_map = get_license_skus(client)
+    mfa_map = get_mfa_status(client)
+    print(f"  Resolved {len(sku_map)} license SKUs")
+
+    with open(OUTPUT_FILE, "w", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=FIELDS + ["licenseSkus", "mfaRegistered", "passwordlessCapable", "lastSignIn"])
+        writer.writeheader()
+
+        for u in users:
+            upn = u.user_principal_name or ""
+            mfa = mfa_map.get(upn.upper())
+            last_signin = u.properties.get("signInActivity")
+            if hasattr(last_signin, "last_sign_in_date_time"):
+                last_signin = last_signin.last_sign_in_date_time
+            else:
+                last_signin = u.properties.get("signInActivity/lastSignInDateTime", "")
+
+            licenses = []
+            if u.assigned_licenses:
+                for lic in u.assigned_licenses:
+                    sku_id = lic.properties.get("skuId", "")
+                    licenses.append(sku_map.get(sku_id, sku_id))
+
+            writer.writerow({
+                "userPrincipalName": upn,
+                "displayName": u.display_name or "",
+                "givenName": u.given_name or "",
+                "surname": u.surname or "",
+                "department": u.department or "",
+                "jobTitle": u.job_title or "",
+                "officeLocation": u.office_location or "",
+                "city": u.city or "",
+                "country": u.country or "",
+                "mobilePhone": u.mobile_phone or "",
+                "businessPhones": ", ".join(u.business_phones) if u.business_phones else "",
+                "mail": u.mail or "",
+                "accountEnabled": u.account_enabled,
+                "userType": u.user_type or "",
+                "createdDateTime": u.created_date_time.strftime("%Y-%m-%d") if u.created_date_time else "",
+                "signInActivity": str(last_signin)[:10] if last_signin else "",
+                "assignedLicenses": ";".join(licenses) if licenses else "",
+                "licenseSkus": ";".join(licenses) if licenses else "",
+                "mfaRegistered": str(mfa.is_mfa_registered) if mfa else "",
+                "passwordlessCapable": str(mfa.is_passwordless_capable) if mfa else "",
+                "lastSignIn": str(last_signin)[:10] if last_signin else "",
+            })
+
+    print(f"\n✓ Exported {len(users)} users to {OUTPUT_FILE}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/examples/entraid/migration/import_users_csv.py b/examples/entraid/migration/import_users_csv.py
@@ -0,0 +1,96 @@
+"""
+Import users from CSV — bulk provision users from a structured file.
+
+Reads a CSV with user properties and creates each user in Entra ID.
+Useful for:
+  - Tenant-to-tenant migration user provisioning
+  - HR system integration (new hire provisioning)
+  - Bulk user creation for testing or staging environments
+
+CSV format:
+    userPrincipalName,givenName,surname,displayName,department,jobTitle,
+    officeLocation,city,country,mobilePhone,mail,password,accountEnabled
+
+Requires delegated permission ``User.ReadWrite.All``.
+
+https://learn.microsoft.com/en-us/graph/api/user-post-users
+"""
+
+import csv
+import sys
+from io import StringIO
+
+from office365.graph_client import GraphClient
+from tests import create_unique_name, test_client_id, test_client_secret, test_tenant
+
+SAMPLE_CSV = """userPrincipalName,givenName,surname,displayName,department,jobTitle,officeLocation,city,country,mobilePhone,mail,password,accountEnabled
+user1@contoso.com,John,Doe,John Doe,Engineering,Developer,Seattle,Seattle,USA,+1-555-0101,,TempP@ss123,True
+user2@contoso.com,Jane,Smith,Jane Smith,Marketing,Manager,New York,New York,USA,+1-555-0102,,TempP@ss456,True
+user3@contoso.com,Bob,Johnson,Bob Johnson,Sales,Representative,Chicago,Chicago,USA,+1-555-0103,,TempP@ss789,True
+"""
+
+
+def import_users_from_csv(csv_content: str) -> list[dict]:
+    """Import users from CSV content.
+
+    Returns list of result dicts with status per user.
+    """
+    client = GraphClient(tenant=test_tenant).with_client_secret(test_client_id, test_client_secret)
+    reader = csv.DictReader(StringIO(csv_content))
+    results = []
+
+    for row in reader:
+        upn = row["userPrincipalName"].strip()
+        password = row.get("password", "").strip() or create_unique_name("P@ssw0rd")
+
+        try:
+            # Build user profile
+            user = client.users.add_with_password(
+                userPrincipalName=upn,
+                givenName=row.get("givenName", "").strip(),
+                surname=row.get("surname", "").strip(),
+                displayName=row.get("displayName", "").strip() or row.get("givenName", "").strip(),
+                department=row.get("department", "").strip(),
+                jobTitle=row.get("jobTitle", "").strip(),
+                officeLocation=row.get("officeLocation", "").strip(),
+                city=row.get("city", "").strip(),
+                country=row.get("country", "").strip(),
+                mobilePhone=row.get("mobilePhone", "").strip(),
+                mail=row.get("mail", "").strip() or upn,
+                password=password,
+                accountEnabled=row.get("accountEnabled", "True").strip().lower() == "true",
+            ).execute_query()
+
+            results.append({"user": upn, "status": "success", "id": user.id})
+        except Exception as e:
+            results.append({"user": upn, "status": "failed", "error": str(e)})
+
+    return results
+
+
+def main():
+    print("Importing users from CSV\n")
+
+    # In production, read from a file:
+    # with open("users.csv") as f:
+    #     csv_content = f.read()
+    csv_content = SAMPLE_CSV
+
+    results = import_users_from_csv(csv_content)
+
+    success = [r for r in results if r["status"] == "success"]
+    failed = [r for r in results if r["status"] == "failed"]
+
+    print(f"Total:   {len(results)}")
+    print(f"Created: {len(success)}")
+    print(f"Failed:  {len(failed)}\n")
+
+    for r in failed:
+        print(f"  ✗ {r['user']}: {r.get('error', 'Unknown')}")
+
+    for r in success:
+        print(f"  ✓ {r['user']}  (id: {r['id'][:20]}...)")
+
+
+if __name__ == "__main__":
+    main()
