feat: add missing TeamsAsyncOperation properties and stubs for OperationError, TeamsAsyncOperationType

Author: vgrem

examples/teams/archive_lifecycle.py

@@ -1,135 +1,44 @@
 """
-Archive lifecycle: find inactive teams, notify owners, and archive
-(or later unarchive on demand).
+Find inactive teams and report archiving candidates.
 
-Archived teams are read-only — no new messages, edits, or reactions.
-They remain in the directory and can be restored at any time.
-This is a common compliance workflow for reducing team sprawl.
-
-The script operates in two modes:
-  - **audit** (default): scan teams, report candidates for archiving.
-  - **archive**: archive teams inactive for the given threshold.
-
-Requires application permissions:
-    Team.ReadBasic.All, TeamSettings.Read.All, TeamMember.Read.All
-    Directory.Read.All               List all teams
-    TeamSettings.ReadWrite.All       Archive / unarchive
-
-https://learn.microsoft.com/en-us/graph/api/team-archive
-https://learn.microsoft.com/en-us/graph/api/team-unarchive
+Archived teams are read-only. Requires application permissions
+Team.ReadBasic.All, TeamMember.Read.All, Directory.Read.All.
 """
 
-import sys
 from datetime import datetime, timedelta, timezone
 
 from office365.graph_client import GraphClient
 from tests import test_client_id, test_client_secret, test_tenant
 
-INACTIVITY_DAYS = 180  # Archive teams with no recent channel activity
-
-
-def get_all_teams(client: GraphClient) -> list[dict]:
-    """List every team in the tenant with basic metadata."""
-    groups = (
-        client.groups.get()
-        .filter("resourceProvisioningOptions/Any(x:x eq 'Team')")
-        .select(["id", "displayName", "description", "visibility"])
-        .top(999)
-        .execute_query()
-    )
-    teams = []
-    for g in groups:
-        team = client.teams[g.id].get().select(["isArchived", "createdDateTime"]).execute_query()
-        teams.append(
-            {
-                "id": g.id,
-                "name": g.display_name,
-                "description": getattr(g, "description", ""),
-                "visibility": getattr(g, "visibility", "?"),
-                "is_archived": team.is_archived,
-                "created": team.created_date_time,
-            }
-        )
-    return teams
-
-
-def last_activity_date(client: GraphClient, team_id: str) -> datetime | None:
-    """Find the most recent message across any channel in the team.
-
-    Returns None if the team has no messages at all.
-    """
-    latest = None
-    channels = client.teams[team_id].channels.get().execute_query()
-    for channel in channels:
-        try:
-            messages = client.teams[team_id].channels[channel.id].messages.top(1).get().execute_query()
-            if messages:
-                msg_date = getattr(messages[0], "created_date_time", None)
-                if msg_date and (latest is None or msg_date > latest):
-                    latest = msg_date
-        except Exception:
-            pass
-    return latest
+INACTIVITY_DAYS = 180
 
 
 def main():
-    mode = sys.argv[1] if len(sys.argv) > 1 else "audit"
-
     client = GraphClient(tenant=test_tenant).with_client_secret(test_client_id, test_client_secret)
     cutoff = datetime.now(timezone.utc) - timedelta(days=INACTIVITY_DAYS)
 
-    teams = get_all_teams(client)
-    print(f"Found {len(teams)} teams in the tenant\n")
-
-    candidates = []
-    for t in teams:
-        if t["is_archived"]:
-            continue  # already archived
-        last_msg = last_activity_date(client, t["id"])
-        if last_msg is None or last_msg < cutoff:
-            days = (datetime.now(timezone.utc) - last_msg).days if last_msg else float("inf")
-            candidates.append({**t, "last_message": last_msg, "days_since": days})
-
-    if not candidates:
-        print("No inactive teams found (all have recent activity).")
-        return
-
-    print(f"Inactive teams ({INACTIVITY_DAYS}+ days): {len(candidates)}\n")
-    print(f"{'Team':35s} {'Visibility':12s} {'Created':15s} {'Days inactive':>14s} {'Archived?'}")
-    print("-" * 95)
+    teams = client.teams.get_all().execute_query()
+    print(f"Found {len(teams)} teams\n")
 
-    for c in sorted(candidates, key=lambda x: x["days_since"], reverse=True):
-        last_str = f"{c['days_since']:.0f}d" if c["last_message"] else "∞ (no messages)"
-        created_str = c["created"].strftime("%Y-%m-%d") if c["created"] else "?"
-        print(
-            f"{c['name'][:33]:35s} "
-            f"{c['visibility']:12s} "
-            f"{created_str:15s} "
-            f"{last_str:>14s}  "
-            f"{'→ will archive' if mode == 'archive' else 'candidate'}"
-        )
+    for team in teams:
+        if team.is_archived:
+            continue
 
-    # — Archive mode —
-    if mode == "archive":
-        print(f"\nArchiving {len(candidates)} teams (set-readonly)...")
-        archived_count = 0
-        for c in candidates:
+        last_msg = None
+        for channel in team.channels.get().execute_query():
             try:
-                # Set "shouldSetSpoSiteReadOnlyForMembers=False" to let
-                # members still access SharePoint even though Teams is archived.
-                client.teams[c["id"]].archive().execute_query()
-                archived_count += 1
-                print(f"  ✓ {c['name']}")
-            except Exception as e:
-                print(f"  ⚠ {c['name']}: {e}")
-        print(f"\nDone. {archived_count} teams archived.")
+                msgs = channel.messages.top(1).order_by("createdDateTime desc").get().execute_query()
+                if msgs:
+                    msg_date = msgs[0].created_date_time
+                    if msg_date and (last_msg is None or msg_date > last_msg):
+                        last_msg = msg_date
+            except Exception:
+                pass
 
-    # — Provide a ready-to-paste unarchive snippet —
-    print()
-    print("To unarchive a specific team later:")
-    print('  team = client.teams["<teamId>"].get().execute_query()')
-    print("  team.unarchive().execute_query()")
-    print('  print(f"Unarchived: {team.display_name}")')
+        if last_msg is None or last_msg < cutoff:
+            days = (datetime.now(timezone.utc) - last_msg).days if last_msg else float("inf")
+            status = f"{days:.0f}d" if last_msg else "no messages"
+            print(f"  {team.display_name}  ({team.visibility})  inactive: {status}")
 
 
 if __name__ == "__main__":

examples/teams/call_records.py

@@ -1,15 +1,7 @@
 """
 Call records — Teams call quality analytics across the tenant.
 
-Rich data model: call → sessions → segments → endpoints.
-Useful for:
-  - Identifying poor-quality calls (jitter, packet loss, latency)
-  - Auditing direct routing call failures
-  - Reporting on call modalities (audio, video, screen share)
-
-Requires application permission ``CallRecords.Read.All``.
-
-https://learn.microsoft.com/en-us/graph/api/resources/callrecords-api-overview
+Requires application permission CallRecords.Read.All.
 """
 
 from office365.graph_client import GraphClient
@@ -19,72 +11,27 @@
 def main():
     client = GraphClient(tenant=test_tenant).with_client_secret(test_client_id, test_client_secret)
 
-    # -- Step 1: list recent call records --
-    records = client.communications.call_records.top(20).get().execute_query()
+    records = client.communications.call_records.get().execute_query()
     print(f"Call records: {len(records)}\n")
 
-    if not records:
-        print("(No call records — may need CallRecords.Read.All permission or recent calls)")
-        return
-
     for r in records:
-        r_id = r.id or "?"
-        call_type = r.properties.get("type", "?")
-        modality = r.properties.get("modalities", [])
-        start = r.start_date_time
-        if hasattr(start, "strftime"):
-            start = start.strftime("%Y-%m-%d %H:%M")
-        end = r.end_date_time
-        if hasattr(end, "strftime"):
-            end = end.strftime("%Y-%m-%d %H:%M")
-
-        user_count = r.properties.get("participants_count", "?")
-        organizer = r.properties.get("organizer", {})
-        org_upn = "?"
-        if isinstance(organizer, dict):
-            u = organizer.get("user", {})
-            org_upn = u.get("userPrincipalName", "?") if isinstance(u, dict) else "?"
-
-        failure = r.properties.get("failureInfo", {})
-        failure_phase = failure.get("stage", "") if isinstance(failure, dict) else ""
+        org = r.organizer.user.user_principal_name if r.organizer and r.organizer.user else "?"
+        print(f"  {r.start_date_time}  {r.type_.name}  organizer={org}")
 
-        print(
-            f"  {r_id[:20]:20s}  type={call_type:15s}  modality={str(modality[:3]):10s}  "
-            f"users={user_count}  organizer={org_upn[:25]:25s}"
-        )
-        print(f"    start={start}  end={end}  failure={failure_phase or 'none'}")
-
-    # -- Step 2: drill into sessions for the most recent call --
     if records:
         print()
         recent = records[0]
-        print(f"Drilling into call {recent.id[:20]}...")
-
         sessions = recent.sessions.get().execute_query()
-        print(f"  Sessions: {len(sessions)}")
-
+        print(f"Sessions: {len(sessions)}")
         for s in sessions:
-            s_id = s.id[:15] if s.id else "?"
-            callee = s.properties.get("callee", {}).get("user", {}).get("userPrincipalName", "?")
-            caller = s.properties.get("caller", {}).get("user", {}).get("userPrincipalName", "?")
-            modality = s.properties.get("modalities", [])
-            print(
-                f"    session {s_id}  caller={str(caller)[:25]:25s}  "
-                f"callee={str(callee)[:25]:25s}  modality={str(modality[:3]):10s}"
-            )
+            caller = s.caller.user.user_principal_name if s.caller and s.caller.user else "?"
+            callee = s.callee.user.user_principal_name if s.callee and s.callee.user else "?"
+            modalities = [m.name for m in s.modalities]
+            print(f"  {caller} -> {callee}  modalities={modalities}")
 
-            # Segments within this session
             segments = s.segments.get().execute_query()
             for seg in segments:
-                seg_id = seg.properties.get("id", "?")[:10]
-                print(f"      segment {seg_id}")
-
-                # Endpoint quality data
-                endpoints = seg.properties.get("endpoints", [])
-                for ep in endpoints:
-                    name = ep.get("name", "?")
-                    ep_type = ep.get("@odata.type", "?")
-                    print(f"        endpoint={name}  type={str(ep_type).split('.')[-1][:20]}")
+                print(f"    segment {seg.id}")
 
 
 if __name__ == "__main__":

examples/teams/clone_team.py

@@ -59,15 +59,7 @@ def main():
         f"{source_team.display_name}_cloned",
         ClonableTeamParts.channels,
         source_team.visibility,
-    ).execute_query()
-
-    # — Step 4: poll for completion via the source team's operations —
-    succeeded = poll_clone_completion(source_team)
-
-    if succeeded:
-        print(f"\n Clone succeeded — '{target_team.display_name}' is ready.")
-    else:
-        print("\n Clone failed or timed out.")
+    ).execute_query_and_wait()
 
 
 if __name__ == "__main__":

office365/communications/callrecords/session.py

@@ -8,6 +8,7 @@
 from office365.entity_collection import EntityCollection
 from office365.runtime.client_value_collection import ClientValueCollection
 from office365.runtime.paths.resource_path import ResourcePath
+from office365.runtime.types.odata_property import odata
 
 
 class Session(Entity):
@@ -32,6 +33,7 @@ def caller(self) -> Endpoint:
         """Gets the caller property"""
         return self.properties.get("caller", Endpoint())
 
+    @odata(name="endDateTime")
     @property
     def end_date_time(self) -> datetime:
         """Gets the endDateTime property"""
@@ -47,6 +49,7 @@ def modalities(self) -> ClientValueCollection[Modality]:
         """Gets the modalities property"""
         return self.properties.get("modalities", ClientValueCollection[Modality](Modality))
 
+    @odata(name="startDateTime")
     @property
     def start_date_time(self) -> datetime:
         """Gets the startDateTime property"""

office365/teams/collection.py

@@ -55,6 +55,7 @@ def _process_response(resp: requests.Response) -> None:
             operation_path = ODataPathBuilder.parse_url(loc)
             operation = TeamsAsyncOperation(self.context, operation_path)
             return_type.operations.add_child(operation)
+            return_type._pending_operation = operation
 
         payload = {
             "displayName": display_name,

office365/teams/operations/async_operation.py

@@ -1,7 +1,11 @@
 import time
+from datetime import datetime
 from typing import Optional
 
 from office365.entity import Entity
+from office365.runtime.types.odata_property import odata
+from office365.teams.operations.error import OperationError
+from office365.teams.operations.type import TeamsAsyncOperationType
 
 
 class TeamsAsyncOperation(Entity):
@@ -54,6 +58,40 @@ def _verify_status(return_type):
         _poll_for_status(1)
         return self
 
+    @odata(name="attemptsCount")
+    @property
+    def attempts_count(self) -> Optional[int]:
+        """Number of times the operation was attempted before being marked as succeeded or failed."""
+        return self.properties.get("attemptsCount", None)
+
+    @odata(name="createdDateTime")
+    @property
+    def created_date_time(self) -> Optional[datetime]:
+        """Date and time when the operation was created."""
+        return self.properties.get("createdDateTime", None)
+
+    @property
+    def error(self) -> Optional[OperationError]:
+        """Error information if the operation failed."""
+        return self.properties.get("error", None)
+
+    @odata(name="lastActionDateTime")
+    @property
+    def last_action_date_time(self) -> Optional[datetime]:
+        """Date and time when the operation was last updated."""
+        return self.properties.get("lastActionDateTime", None)
+
+    @odata(name="operationType")
+    @property
+    def operation_type(self) -> Optional[TeamsAsyncOperationType]:
+        """The type of the operation."""
+        return self.properties.get("operationType", None)
+
+    @property
+    def status(self) -> Optional[str]:
+        """Operation status."""
+        return self.properties.get("status", None)
+
     @property
     def target_resource_id(self) -> Optional[str]:
         """The ID of the object that's created or modified as result of this async operation, typically a team."""
@@ -65,8 +103,3 @@ def target_resource_location(self) -> Optional[str]:
         This URL should be treated as an opaque value and not parsed into its component paths.
         """
         return self.properties.get("targetResourceLocation", None)
-
-    @property
-    def status(self) -> Optional[str]:
-        """Operation status."""
-        return self.properties.get("status", None)

office365/teams/operations/error.py

@@ -0,0 +1,5 @@
+from office365.runtime.client_value import ClientValue
+
+
+class OperationError(ClientValue):
+    """Represents the error information for a failed Teams async operation."""

office365/teams/operations/type.py

@@ -0,0 +1,16 @@
+from enum import Enum
+
+
+class TeamsAsyncOperationType(Enum):
+    """The type of long-running operation for a team."""
+
+    invalid = "invalid"
+    cloneTeam = "cloneTeam"
+    archiveTeam = "archiveTeam"
+    unarchiveTeam = "unarchiveTeam"
+    createTeam = "createTeam"
+    unknownFutureValue = "unknownFutureValue"
+
+    @property
+    def entity_type_name(self):
+        return "microsoft.graph.teamsAsyncOperationType"