Remove dependence on pytz

Author: kroenlein

requirements.txt

@@ -3,7 +3,6 @@ boto3==1.34.35
 deprecation==2.1.0
 gemd==2.1.9
 pyjwt==2.8.0
-pytz==2024.1; python_version < "3.9"
 requests==2.32.2
 tqdm==4.66.3
 

setup.py

@@ -30,7 +30,6 @@
           "deprecation>=2.1.0,<3",
           "urllib3>=1.26.18,<3",
           "tqdm>=4.27.0,<5",
-          "pytz>=2024.1; python_version<'3.9'",
       ],
       extras_require={
           "tests": [

src/citrine/_session.py

@@ -1,9 +1,5 @@
 import platform
-from datetime import datetime, timedelta
-try:  # Only available starting Python 3.9
-    from datetime import UTC
-except ImportError:  # pragma: no cover
-    from pytz import utc as UTC
+from datetime import datetime, timedelta, timezone
 from json.decoder import JSONDecodeError
 from logging import getLogger
 from os import environ
@@ -29,7 +25,7 @@
 
 # Choose a 5-second buffer so that there's no chance of the access token
 # expiring during the check for expiration
-EXPIRATION_BUFFER_MILLIS: timedelta = timedelta(milliseconds=5000)
+EXPIRATION_BUFFER: timedelta = timedelta(seconds=5)
 logger = getLogger(__name__)
 
 
@@ -57,7 +53,7 @@ def __init__(self,
         self.authority = ':'.join(([host] if host else []) + ([port] if port else []))
         self.refresh_token: str = refresh_token
         self.access_token: Optional[str] = None
-        self.access_token_expiration: datetime = datetime.now(UTC)
+        self.access_token_expiration: datetime = datetime.now(timezone.utc)
 
         agent = "{}/{} python-requests/{} citrine-python/{}".format(
             platform.python_implementation(),
@@ -110,7 +106,8 @@ def _versioned_base_url(self, version: str = 'v1'):
         ))
 
     def _is_access_token_expired(self):
-        return self.access_token_expiration - EXPIRATION_BUFFER_MILLIS <= datetime.now(UTC)
+        buffered_expire = self.access_token_expiration - EXPIRATION_BUFFER
+        return datetime.now(timezone.utc) > buffered_expire
 
     def _refresh_access_token(self) -> None:
         """Optionally refresh our access token (if the previous one is about to expire)."""
@@ -128,7 +125,7 @@ def _refresh_access_token(self) -> None:
                 options={"verify_signature": False},
                 algorithms=["HS256"]
             )['exp'],
-            UTC
+            timezone.utc
         )
 
         # Explicitly set an updated 'auth', so as to not rely on implicit cookie handling.

tests/test_citrine.py

@@ -1,9 +1,5 @@
 import platform
-from datetime import datetime
-try:
-    from datetime import UTC
-except ImportError:  # Only available starting Python 3.9
-    from pytz import utc as UTC
+from datetime import datetime, timezone
 
 import jwt
 import pytest
@@ -20,7 +16,7 @@ def refresh_token(expiration: datetime = None) -> dict:
     return {'access_token': token}
 
 
-token_refresh_response = refresh_token(datetime(2019, 3, 14, tzinfo=UTC))
+token_refresh_response = refresh_token(datetime(2019, 3, 14, tzinfo=timezone.utc))
 
 
 def test_citrine_creation():

tests/test_session.py

@@ -8,11 +8,7 @@
     WorkflowNotReadyException,
     RetryableException)
 
-from datetime import datetime, timedelta
-try:
-    from datetime import UTC
-except ImportError:  # Only available starting Python 3.9
-    from pytz import utc as UTC
+from datetime import datetime, timedelta, timezone
 
 import mock
 import requests
@@ -32,7 +28,7 @@ def refresh_token(expiration: datetime = None) -> dict:
 
 @pytest.fixture
 def session():
-    token_refresh_response = refresh_token(datetime(2019, 3, 14, tzinfo=UTC))
+    token_refresh_response = refresh_token(datetime(2019, 3, 14, tzinfo=timezone.utc))
     with requests_mock.Mocker() as m:
         m.post('http://citrine-testing.fake/api/v1/tokens/refresh', json=token_refresh_response)
         session = Session(
@@ -43,13 +39,13 @@ def session():
     # Default behavior is to *not* require a refresh - those tests can clear this out
     # As rule of thumb, we should be using freezegun or similar to never rely on the system clock
     # for these scenarios, but I thought this is light enough to postpone that for the time being
-    session.access_token_expiration = datetime.now(UTC) + timedelta(minutes=3)
+    session.access_token_expiration = datetime.now(timezone.utc) + timedelta(minutes=3)
 
     return session
 
 
 def test_session_signature(monkeypatch):
-    token_refresh_response = refresh_token(datetime(2019, 3, 14, tzinfo=UTC))
+    token_refresh_response = refresh_token(datetime(2019, 3, 14, tzinfo=timezone.utc))
     with requests_mock.Mocker() as m:
         m.post('ftp://citrine-testing.fake:8080/api/v1/tokens/refresh', json=token_refresh_response)
 
@@ -77,8 +73,8 @@ def test_session_signature(monkeypatch):
 
 
 def test_get_refreshes_token(session: Session):
-    session.access_token_expiration = datetime.now(UTC) - timedelta(minutes=1)
-    token_refresh_response = refresh_token(datetime(2019, 3, 14, tzinfo=UTC))
+    session.access_token_expiration = datetime.now(timezone.utc) - timedelta(minutes=1)
+    token_refresh_response = refresh_token(datetime(2019, 3, 14, tzinfo=timezone.utc))
 
     with requests_mock.Mocker() as m:
         m.post('http://citrine-testing.fake/api/v1/tokens/refresh', json=token_refresh_response)
@@ -89,11 +85,11 @@ def test_get_refreshes_token(session: Session):
         resp = session.get_resource('/foo')
 
     assert {'foo': 'bar'} == resp
-    assert datetime(2019, 3, 14, tzinfo=UTC) == session.access_token_expiration
+    assert datetime(2019, 3, 14, tzinfo=timezone.utc) == session.access_token_expiration
 
 
 def test_get_refresh_token_failure(session: Session):
-    session.access_token_expiration = datetime.now(UTC) - timedelta(minutes=1)
+    session.access_token_expiration = datetime.now(timezone.utc) - timedelta(minutes=1)
 
     with requests_mock.Mocker() as m:
         m.post('http://citrine-testing.fake/api/v1/tokens/refresh', status_code=401)
@@ -197,7 +193,7 @@ def test_connection_error(session: Session):
 
 
 def test_post_refreshes_token_when_denied(session: Session):
-    token_refresh_response = refresh_token(datetime(2019, 3, 14, tzinfo=UTC))
+    token_refresh_response = refresh_token(datetime(2019, 3, 14, tzinfo=timezone.utc))
 
     with requests_mock.Mocker() as m:
         m.post('http://citrine-testing.fake/api/v1/tokens/refresh', json=token_refresh_response)
@@ -209,7 +205,7 @@ def test_post_refreshes_token_when_denied(session: Session):
         resp = session.post_resource('/foo', json={'data': 'hi'})
 
     assert {'foo': 'bar'} == resp
-    assert datetime(2019, 3, 14, tzinfo=UTC) == session.access_token_expiration
+    assert datetime(2019, 3, 14, tzinfo=timezone.utc) == session.access_token_expiration
 
 
 # this test exists to provide 100% coverage for the legacy 401 status on Unauthorized responses