fix: move away from deprecated datetime utilities

Author: WilliamBergamin

slack_sdk/oauth/installation_store/models/bot.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timezone
 from time import time
 from typing import Optional, Union, Dict, Any, Sequence
 
@@ -100,10 +100,12 @@ def _to_standard_value_dict(self) -> Dict[str, Any]:
             "bot_scopes": ",".join(self.bot_scopes) if self.bot_scopes else None,
             "bot_refresh_token": self.bot_refresh_token,
             "bot_token_expires_at": (
-                datetime.utcfromtimestamp(self.bot_token_expires_at) if self.bot_token_expires_at is not None else None
+                datetime.fromtimestamp(self.bot_token_expires_at, tz=timezone.utc)
+                if self.bot_token_expires_at is not None
+                else None
             ),
             "is_enterprise_install": self.is_enterprise_install,
-            "installed_at": datetime.utcfromtimestamp(self.installed_at),
+            "installed_at": datetime.fromtimestamp(self.installed_at, tz=timezone.utc),
         }
 
     def to_dict_for_copying(self) -> Dict[str, Any]:

slack_sdk/oauth/installation_store/models/installation.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timezone
 from time import time
 from typing import Optional, Union, Dict, Any, Sequence
 
@@ -173,22 +173,26 @@ def _to_standard_value_dict(self) -> Dict[str, Any]:
             "bot_scopes": ",".join(self.bot_scopes) if self.bot_scopes else None,
             "bot_refresh_token": self.bot_refresh_token,
             "bot_token_expires_at": (
-                datetime.utcfromtimestamp(self.bot_token_expires_at) if self.bot_token_expires_at is not None else None
+                datetime.fromtimestamp(self.bot_token_expires_at, tz=timezone.utc)
+                if self.bot_token_expires_at is not None
+                else None
             ),
             "user_id": self.user_id,
             "user_token": self.user_token,
             "user_scopes": ",".join(self.user_scopes) if self.user_scopes else None,
             "user_refresh_token": self.user_refresh_token,
             "user_token_expires_at": (
-                datetime.utcfromtimestamp(self.user_token_expires_at) if self.user_token_expires_at is not None else None
+                datetime.fromtimestamp(self.user_token_expires_at, tz=timezone.utc)
+                if self.user_token_expires_at is not None
+                else None
             ),
             "incoming_webhook_url": self.incoming_webhook_url,
             "incoming_webhook_channel": self.incoming_webhook_channel,
             "incoming_webhook_channel_id": self.incoming_webhook_channel_id,
             "incoming_webhook_configuration_url": self.incoming_webhook_configuration_url,
             "is_enterprise_install": self.is_enterprise_install,
             "token_type": self.token_type,
-            "installed_at": datetime.utcfromtimestamp(self.installed_at),
+            "installed_at": datetime.fromtimestamp(self.installed_at, tz=timezone.utc),
         }
 
     def to_dict_for_copying(self) -> Dict[str, Any]:

slack_sdk/oauth/state_store/sqlalchemy/__init__.py

@@ -1,6 +1,6 @@
 import logging
 import time
-from datetime import datetime
+from datetime import datetime, timezone
 from logging import Logger
 from uuid import uuid4
 
@@ -55,7 +55,7 @@ def logger(self) -> Logger:
 
     def issue(self, *args, **kwargs) -> str:
         state: str = str(uuid4())
-        now = datetime.utcfromtimestamp(time.time() + self.expiration_seconds)
+        now = datetime.fromtimestamp(time.time() + self.expiration_seconds, tz=timezone.utc)
         with self.engine.begin() as conn:
             conn.execute(
                 self.oauth_states.insert(),
@@ -67,7 +67,7 @@ def consume(self, state: str) -> bool:
         try:
             with self.engine.begin() as conn:
                 c = self.oauth_states.c
-                query = self.oauth_states.select().where(and_(c.state == state, c.expire_at > datetime.utcnow()))
+                query = self.oauth_states.select().where(and_(c.state == state, c.expire_at > datetime.now(tz=timezone.utc)))
                 result = conn.execute(query)
                 for row in result.mappings():
                     self.logger.debug(f"consume's query result: {row}")
@@ -124,7 +124,7 @@ def logger(self) -> Logger:
 
     async def async_issue(self, *args, **kwargs) -> str:
         state: str = str(uuid4())
-        now = datetime.utcfromtimestamp(time.time() + self.expiration_seconds)
+        now = datetime.fromtimestamp(time.time() + self.expiration_seconds, tz=timezone.utc)
         async with self.engine.begin() as conn:
             await conn.execute(
                 self.oauth_states.insert(),
@@ -136,7 +136,7 @@ async def async_consume(self, state: str) -> bool:
         try:
             async with self.engine.begin() as conn:
                 c = self.oauth_states.c
-                query = self.oauth_states.select().where(and_(c.state == state, c.expire_at > datetime.utcnow()))
+                query = self.oauth_states.select().where(and_(c.state == state, c.expire_at > datetime.now(tz=timezone.utc)))
                 result = await conn.execute(query)
                 for row in result.mappings():
                     self.logger.debug(f"consume's query result: {row}")

tests/slack_sdk/oauth/installation_store/test_models.py

@@ -1,4 +1,5 @@
 import time
+from datetime import datetime, timezone
 import unittest
 
 from slack_sdk.oauth.installation_store import Installation, FileInstallationStore, Bot
@@ -36,6 +37,22 @@ def test_bot_custom_fields(self):
         self.assertEqual(bot.to_dict().get("service_user_id"), "XYZ123")
         self.assertEqual(bot.to_dict_for_copying().get("custom_values").get("service_user_id"), "XYZ123")
 
+    def test_bot_datetime_manipulation(self):
+        expected_timestamp = datetime.now(tz=timezone.utc)
+        bot = Bot(
+            bot_token="xoxb-",
+            bot_id="B111",
+            bot_user_id="U111",
+            bot_token_expires_at=expected_timestamp,
+            installed_at=expected_timestamp,
+        )
+        bot_dict = bot.to_dict()
+        self.assertIsNotNone(bot_dict)
+        self.assertEqual(
+            bot_dict.get("bot_token_expires_at").isoformat(), expected_timestamp.strftime("%Y-%m-%dT%H:%M:%S+00:00")
+        )
+        self.assertEqual(bot_dict.get("installed_at"), expected_timestamp)
+
     def test_installation(self):
         installation = Installation(
             app_id="A111",
@@ -84,3 +101,29 @@ def test_installation_custom_fields(self):
         self.assertEqual(bot.to_dict().get("app_id"), "A111")
         self.assertEqual(bot.to_dict().get("service_user_id"), "XYZ123")
         self.assertEqual(bot.to_dict_for_copying().get("custom_values").get("app_id"), "A222")
+
+    def test_installation_datetime_manipulation(self):
+        expected_timestamp = datetime.now(tz=timezone.utc)
+        installation = Installation(
+            app_id="A111",
+            enterprise_id="E111",
+            team_id="T111",
+            user_id="U111",
+            bot_id="B111",
+            bot_token="xoxb-111",
+            bot_scopes=["chat:write"],
+            bot_user_id="U222",
+            bot_token_expires_at=expected_timestamp,
+            user_token_expires_at=expected_timestamp,
+            installed_at=expected_timestamp,
+        )
+        installation_dict = installation.to_dict()
+        self.assertIsNotNone(installation_dict)
+        self.assertEqual(
+            installation_dict.get("bot_token_expires_at").isoformat(), expected_timestamp.strftime("%Y-%m-%dT%H:%M:%S+00:00")
+        )
+        self.assertEqual(
+            installation_dict.get("user_token_expires_at").isoformat(),
+            expected_timestamp.strftime("%Y-%m-%dT%H:%M:%S+00:00"),
+        )
+        self.assertEqual(installation_dict.get("installed_at"), expected_timestamp)