From 9aeb7789d300015f2467dd0f5f456b165c84be18 Mon Sep 17 00:00:00 2001 From: Aaron Bach Date: Fri, 12 Jan 2024 17:05:13 -0700 Subject: [PATCH] Address upcoming deprecation of `datetime.datetime.utcnow()` (#719) * Address upcoming deprecation of `datetime.datetime.utcnow()` * Support Python 3.10 --- pyproject.toml | 4 ++-- simplipy/api.py | 5 +++-- simplipy/system/v3.py | 5 +++-- simplipy/util/dt.py | 21 +++++++++++++++++++-- simplipy/websocket.py | 4 ++-- tests/system/test_v3.py | 3 ++- tests/test_api.py | 9 +++++---- tests/test_lock.py | 5 +++-- 8 files changed, 39 insertions(+), 17 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 921cfc88..dc31c8c8 100644 --- a/pyproject.toml +++ b/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 diff --git a/simplipy/api.py b/simplipy/api.py index 277d2855..63de352e 100644 --- a/simplipy/api.py +++ b/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 diff --git a/simplipy/system/v3.py b/simplipy/system/v3.py index 3f50b410..118590c4 100644 --- a/simplipy/system/v3.py +++ b/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 diff --git a/simplipy/util/dt.py b/simplipy/util/dt.py index b1aca5bc..995950ba 100644 --- a/simplipy/util/dt.py +++ b/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) diff --git a/simplipy/websocket.py b/simplipy/websocket.py index 90529840..ed8e293a 100644 --- a/simplipy/websocket.py +++ b/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" diff --git a/tests/system/test_v3.py b/tests/system/test_v3.py index f0bb5ad3..48b08dc5 100644 --- a/tests/system/test_v3.py +++ b/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] diff --git a/tests/test_api.py b/tests/test_api.py index c4ab929b..1f59eb2a 100644 --- a/tests/test_api.py +++ b/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) diff --git a/tests/test_lock.py b/tests/test_lock.py index 20528224..771608c3 100644 --- a/tests/test_lock.py +++ b/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])