Address upcoming deprecation of datetime.datetime.utcnow() (#719)

* Address upcoming deprecation of `datetime.datetime.utcnow()`

* Support Python 3.10

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "poetry.core.masonry.api"
 target-version = ["py39"]
 
 [tool.coverage.report]
-exclude_lines = ["raise NotImplementedError", "TYPE_CHECKING"]
+exclude_lines = ["raise NotImplementedError", "TYPE_CHECKING", "ImportError"]
 fail_under = 100
 show_missing = true
 
@@ -30,7 +30,7 @@ follow_imports = "silent"
 ignore_missing_imports = true
 no_implicit_optional = true
 platform = "linux"
-python_version = "3.10"
+python_version = "3.12"
 show_error_codes = true
 strict_equality = true
 warn_incomplete_stub = true

simplipy/api.py

@@ -28,6 +28,7 @@
     DEFAULT_CLIENT_ID,
     DEFAULT_REDIRECT_URI,
 )
+from simplipy.util.dt import utcnow
 from simplipy.websocket import WebsocketClient
 
 API_URL_HOSTNAME = "api.simplisafe.com"
@@ -167,7 +168,7 @@ async def _async_handle_on_backoff(self, _: dict[str, Any]) -> None:
         if err.status == 401 and self._token_last_refreshed:
             # Calculate the window between now and the last time the token was
             # refreshed:
-            window = (datetime.utcnow() - self._token_last_refreshed).total_seconds()
+            window = (utcnow() - self._token_last_refreshed).total_seconds()
 
             # Since we might have multiple requests (each running their own retry
             # sequence) land here, we only refresh the access token if it hasn't
@@ -253,7 +254,7 @@ def _save_token_data_from_response(self, token_data: dict[str, Any]) -> None:
         Args:
             token_data: An API response payload.
         """
-        self._token_last_refreshed = datetime.utcnow()
+        self._token_last_refreshed = utcnow()
         self.access_token = token_data["access_token"]
         if refresh_token := token_data.get("refresh_token"):
             self.refresh_token = refresh_token

simplipy/system/v3.py

@@ -20,6 +20,7 @@
     SystemStates,
     guard_from_missing_data,
 )
+from simplipy.util.dt import utcnow
 
 if TYPE_CHECKING:
     from simplipy.api import API
@@ -404,7 +405,7 @@ async def _async_set_state(self, value: SystemStates) -> None:
         )
 
         self._state = value
-        self._last_state_change_dt = datetime.utcnow()
+        self._last_state_change_dt = utcnow()
 
     async def _async_set_updated_pins(self, pins: dict[str, Any]) -> None:
         """Post new PINs.
@@ -612,7 +613,7 @@ async def async_update(
         if (
             self.locks
             and self._last_state_change_dt
-            and datetime.utcnow()
+            and utcnow()
             <= self._last_state_change_dt + DEFAULT_LOCK_STATE_CHANGE_WINDOW
         ):
             # The SimpliSafe cloud API currently has a bug wherein systems with locks

simplipy/util/dt.py

@@ -1,5 +1,22 @@
 """Define datetime utilities."""
-from datetime import datetime, timezone
+from datetime import datetime
+
+try:
+    from datetime import UTC
+except ImportError:
+    # In place for support of Python 3.10
+    from datetime import timezone
+
+    UTC = timezone.utc
+
+
+def utcnow() -> datetime:
+    """Return the current UTC time.
+
+    Returns:
+        A ``datetime.datetime`` object.
+    """
+    return datetime.now(tz=UTC)
 
 
 def utc_from_timestamp(timestamp: float) -> datetime:
@@ -11,4 +28,4 @@ def utc_from_timestamp(timestamp: float) -> datetime:
     Returns:
         A parsed ``datetime.datetime`` object.
     """
-    return datetime.fromtimestamp(timestamp, tz=timezone.utc)
+    return datetime.fromtimestamp(timestamp, tz=UTC)

simplipy/websocket.py

@@ -20,7 +20,7 @@
     NotConnectedError,
 )
 from simplipy.util import CallbackType, execute_callback
-from simplipy.util.dt import utc_from_timestamp
+from simplipy.util.dt import utc_from_timestamp, utcnow
 
 if TYPE_CHECKING:
     from simplipy import API
@@ -405,7 +405,7 @@ async def async_disconnect(self) -> None:
 
     async def async_listen(self) -> None:
         """Start listening to the websocket server."""
-        now = datetime.utcnow()
+        now = utcnow()
         now_ts = round(now.timestamp() * 1000)
         now_utc_iso = f"{now.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3]}Z"
 

tests/system/test_v3.py

@@ -20,6 +20,7 @@
 )
 from simplipy.system import SystemStates
 from simplipy.system.v3 import SystemV3, Volume
+from simplipy.util.dt import utcnow
 from tests.common import (
     TEST_AUTHORIZATION_CODE,
     TEST_CODE_VERIFIER,
@@ -1134,7 +1135,7 @@ async def test_no_state_change_on_failure(
 
             # pylint: disable=protected-access
             # Manually set the expiration datetime to force a refresh token flow:
-            simplisafe._token_last_refreshed = datetime.utcnow() - timedelta(seconds=30)
+            simplisafe._token_last_refreshed = utcnow() - timedelta(seconds=30)
 
             systems = await simplisafe.async_get_systems()
             system = systems[TEST_SYSTEM_ID]

tests/test_api.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import asyncio
-from datetime import datetime, timedelta
+from datetime import timedelta
 from typing import Any
 from unittest.mock import AsyncMock, Mock, patch
 
@@ -13,6 +13,7 @@
 
 from simplipy import API
 from simplipy.errors import InvalidCredentialsError, RequestError, SimplipyError
+from simplipy.util.dt import utcnow
 
 from .common import (
     TEST_ACCESS_TOKEN,
@@ -87,7 +88,7 @@ async def test_401_refresh_token_failure(
             )
 
             # Manually set the expiration datetime to force a refresh token flow:
-            simplisafe._token_last_refreshed = datetime.utcnow() - timedelta(seconds=30)
+            simplisafe._token_last_refreshed = utcnow() - timedelta(seconds=30)
 
             with pytest.raises(InvalidCredentialsError):
                 await simplisafe.async_get_systems()
@@ -152,7 +153,7 @@ async def test_401_refresh_token_success(
             )
 
             # Manually set the expiration datetime to force a refresh token flow:
-            simplisafe._token_last_refreshed = datetime.utcnow() - timedelta(seconds=30)
+            simplisafe._token_last_refreshed = utcnow() - timedelta(seconds=30)
 
             # If this succeeds without throwing an exception, the retry is successful:
             await simplisafe.async_get_systems()
@@ -397,7 +398,7 @@ async def test_refresh_token_callback(
             )
 
             # Manually set the expiration datetime to force a refresh token flow:
-            simplisafe._token_last_refreshed = datetime.utcnow() - timedelta(seconds=30)
+            simplisafe._token_last_refreshed = utcnow() - timedelta(seconds=30)
 
             # We'll hang onto one callback:
             simplisafe.add_refresh_token_callback(mock_callback_1)

tests/test_lock.py

@@ -2,7 +2,7 @@
 # pylint: disable=protected-access
 from __future__ import annotations
 
-from datetime import datetime, timedelta
+from datetime import timedelta
 from typing import Any, cast
 from unittest.mock import Mock
 
@@ -14,6 +14,7 @@
 from simplipy.device.lock import LockStates
 from simplipy.errors import InvalidCredentialsError
 from simplipy.system.v3 import SystemV3
+from simplipy.util.dt import utcnow
 
 from .common import (
     TEST_AUTHORIZATION_CODE,
@@ -139,7 +140,7 @@ async def test_no_state_change_on_failure(
             )
 
             # Manually set the expiration datetime to force a refresh token flow:
-            simplisafe._token_last_refreshed = datetime.utcnow() - timedelta(seconds=30)
+            simplisafe._token_last_refreshed = utcnow() - timedelta(seconds=30)
 
             systems = await simplisafe.async_get_systems()
             system: SystemV3 = cast(SystemV3, systems[TEST_SYSTEM_ID])