refactor: time-based poll_for_status with failed detection and deferred execution preserved

Author: vgrem

examples/teams/clone_team.py

@@ -1,65 +1,36 @@
 """
 Clone a team — copy its channels, apps, tabs, settings, and/or members.
 
-Uses the ``ClonableTeamParts`` enum as a bitset so you can select
-exactly what to copy.  The clone operation is async; this example
-polls the source team's ``operations`` collection until it completes.
+The clone operation is async; ``execute_query_and_wait`` polls until
+the operation completes.
 
-Typical provisioning pattern: bootstrap a new project from a template
-team that already has the right channel structure and app setup.
-
-Requires delegated permission ``Team.Create`` or ``Group.ReadWrite.All``.
-
-Note: the SDK's ``team.clone()`` helper is a stub that doesn't pass
-the required body parameters yet.  This example uses the underlying
-``ServiceOperationQuery`` directly to send the proper payload.
-
-https://learn.microsoft.com/en-us/graph/api/team-clone
+Requires delegated permission Team.Create or Group.ReadWrite.All.
 """
 
-import sys
-import time
-
 from office365.graph_client import GraphClient
 from office365.teams.clonableteamparts import ClonableTeamParts
-from office365.teams.team import Team
 from tests import test_client_id, test_password, test_tenant, test_username
 
 
-def poll_clone_completion(team: Team, max_wait_sec: int = 120) -> bool:
-    """Poll the team's operations list waiting for a clone to succeed/fail."""
-    for attempt in range(1, (max_wait_sec // 10) + 1):
-        time.sleep(10)
-        ops = team.operations.get().execute_query()
-        for op in ops:
-            print(f"    [{attempt * 10}s] Operation status: {op.status}")
-            if op.status == "succeeded":
-                return True
-            elif op.status == "failed":
-                err = op.properties.get("error", {})
-                print(f"    Error: {err.get('message', 'unknown')}")
-                return False
-    return False
-
-
 def main():
     client = GraphClient(tenant=test_tenant).with_username_and_password(test_client_id, test_username, test_password)
 
-    # — Step 1: pick a source team —
     teams = client.teams.get().top(1).execute_query()
-    if len(teams) == 0:
-        sys.exit("No source team found.")
+    if not teams:
+        print("No source team found.")
+        return
 
     source_team = teams[0]
     print(f"Source team: {source_team.display_name}")
 
-    print("Cloning (this may take 60+ seconds)...\n")
-    target_team = source_team.clone(
+    print("Cloning...")
+    source_team.clone(
         f"{source_team.display_name}_cloned",
         f"{source_team.display_name}_cloned",
         ClonableTeamParts.channels,
         source_team.visibility,
     ).execute_query_and_wait()
+    print("Clone completed.")
 
 
 if __name__ == "__main__":

office365/teams/operations/async_operation.py

@@ -30,38 +30,42 @@ class TeamsAsyncOperation(Entity):
     def poll_for_status(
         self,
         status_type: TeamsAsyncOperationStatus = TeamsAsyncOperationStatus.succeeded,
-        max_polling_count: int = 5,
-        polling_interval_secs: int = 15,
+        timeout_sec: int = 180,
+        polling_interval: int = 15,
         success_callback: Callable[[TeamsAsyncOperation], None] | None = None,
         failure_callback: Callable[[TeamsAsyncOperation], None] | None = None,
     ) -> Self:
-        """Poll to check for completion of an async Teams create call
+        """Poll to check for completion of an async Teams operation.
 
         Args:
-            polling_interval_secs (int):
-            max_polling_count (int):
-            status_type (TeamsAsyncOperationStatus): The status to wait for
-            success_callback ((TeamsAsyncOperation)-> None): Called on success with the operation
-            failure_callback ((TeamsAsyncOperation)-> None): Called on timeout
+            status_type: The status to wait for (default succeeded)
+            timeout_sec: Maximum seconds to wait (default 180)
+            polling_interval: Seconds between polls (default 15)
+            success_callback: Called on success with the populated operation
+            failure_callback: Called on timeout or failed status
         """
+        deadline = time.time() + timeout_sec
 
-        def _poll_for_status(polling_number: int) -> None:
-            if polling_number > max_polling_count:
+        def _poll():
+            self.get().after_execute(_verify_status, execute_first=True)
+
+        def _verify_status(return_type: TeamsAsyncOperation):
+            if return_type.status == status_type:
+                if callable(success_callback):
+                    success_callback(return_type)
+                return
+            if return_type.status == TeamsAsyncOperationStatus.failed:
+                if callable(failure_callback):
+                    failure_callback(return_type)
+                return
+            if time.time() >= deadline:
                 if callable(failure_callback):
                     failure_callback(self)
-                else:
-                    raise TimeoutError("The maximum polling count has been reached")
-
-            def _verify_status(return_type: TeamsAsyncOperation):
-                if return_type.status != status_type:
-                    time.sleep(polling_interval_secs)
-                    _poll_for_status(polling_number + 1)
-                elif callable(success_callback):
-                    success_callback(return_type)
-
-            self.get().after_execute(_verify_status, execute_first=True)
+                return
+            time.sleep(polling_interval)
+            _poll()
 
-        _poll_for_status(1)
+        _poll()
         return self
 
     @odata(name="attemptsCount")