refactor: type-hint poll_for_status with TeamsAsyncOperationStatus, Callable, and Self

Author: vgrem

office365/teams/operations/async_operation.py

@@ -1,9 +1,15 @@
+from __future__ import annotations
+
 import time
+from collections.abc import Callable
 from datetime import datetime
 from typing import Optional
 
+from typing_extensions import Self
+
 from office365.entity import Entity
 from office365.runtime.types.odata_property import odata
+from office365.teams.operations.async_status import TeamsAsyncOperationStatus
 from office365.teams.operations.error import OperationError
 from office365.teams.operations.type import TeamsAsyncOperationType
 
@@ -23,30 +29,30 @@ class TeamsAsyncOperation(Entity):
 
     def poll_for_status(
         self,
-        status_type="succeeded",
-        max_polling_count=5,
-        polling_interval_secs=15,
-        success_callback=None,
-        failure_callback=None,
-    ):
+        status_type: TeamsAsyncOperationStatus = TeamsAsyncOperationStatus.succeeded,
+        max_polling_count: int = 5,
+        polling_interval_secs: 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
 
         Args:
             polling_interval_secs (int):
             max_polling_count (int):
-            status_type (str): The status of a teamsAsyncOperation
-            success_callback ((TeamsAsyncOperation)-> None): A callback to call if the request executes successfully.
-            failure_callback ((TeamsAsyncOperation)-> None): A callback to call if the request fails to execute
+            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
         """
 
         def _poll_for_status(polling_number: int) -> None:
             if polling_number > max_polling_count:
                 if callable(failure_callback):
                     failure_callback(self)
                 else:
-                    raise TypeError("The maximum polling count has been reached")
+                    raise TimeoutError("The maximum polling count has been reached")
 
-            def _verify_status(return_type):
+            def _verify_status(return_type: TeamsAsyncOperation):
                 if return_type.status != status_type:
                     time.sleep(polling_interval_secs)
                     _poll_for_status(polling_number + 1)
@@ -66,31 +72,31 @@ def attempts_count(self) -> Optional[int]:
 
     @odata(name="createdDateTime")
     @property
-    def created_date_time(self) -> Optional[datetime]:
+    def created_date_time(self) -> datetime:
         """Date and time when the operation was created."""
-        return self.properties.get("createdDateTime", None)
+        return self.properties.get("createdDateTime", datetime.min)
 
     @property
-    def error(self) -> Optional[OperationError]:
+    def error(self) -> OperationError:
         """Error information if the operation failed."""
-        return self.properties.get("error", None)
+        return self.properties.get("error", OperationError())
 
     @odata(name="lastActionDateTime")
     @property
-    def last_action_date_time(self) -> Optional[datetime]:
+    def last_action_date_time(self) -> datetime:
         """Date and time when the operation was last updated."""
-        return self.properties.get("lastActionDateTime", None)
+        return self.properties.get("lastActionDateTime", datetime.min)
 
     @odata(name="operationType")
     @property
-    def operation_type(self) -> Optional[TeamsAsyncOperationType]:
+    def operation_type(self) -> TeamsAsyncOperationType:
         """The type of the operation."""
-        return self.properties.get("operationType", None)
+        return self.properties.get("operationType", TeamsAsyncOperationType.unknown)
 
     @property
-    def status(self) -> Optional[str]:
+    def status(self) -> TeamsAsyncOperationStatus:
         """Operation status."""
-        return self.properties.get("status", None)
+        return self.properties.get("status", TeamsAsyncOperationStatus.invalid)
 
     @property
     def target_resource_id(self) -> Optional[str]:

office365/teams/operations/type.py

@@ -4,6 +4,8 @@
 class TeamsAsyncOperationType(Enum):
     """The type of long-running operation for a team."""
 
+    unknown = "unknown"
+
     invalid = "invalid"
     cloneTeam = "cloneTeam"
     archiveTeam = "archiveTeam"

office365/teams/team.py

@@ -24,6 +24,7 @@
 from office365.teams.members.settings import TeamMemberSettings
 from office365.teams.messaging_settings import TeamMessagingSettings
 from office365.teams.operations.async_operation import TeamsAsyncOperation
+from office365.teams.operations.async_status import TeamsAsyncOperationStatus
 from office365.teams.schedule.schedule import Schedule
 from office365.teams.specialization import TeamSpecialization
 from office365.teams.summary import TeamSummary
@@ -53,7 +54,7 @@ def execute_query_and_wait(self) -> Self:
         def _loaded():
             if self._pending_operation is None:
                 raise RuntimeError("No pending async operation to wait for. Call clone() or create() first.")
-            self._pending_operation.poll_for_status(status_type="succeeded")
+            self._pending_operation.poll_for_status(status_type=TeamsAsyncOperationStatus.succeeded)
 
         self.ensure_property("id").after_execute(lambda _: _loaded())
         self.execute_query()