Add token expiration tracking and auto-refresh functionality

- Add is_token_expired() method to check if token needs refresh with configurable buffer
- Add refresh_token_if_needed() for automatic proactive token refresh before expiration
- Add refresh_token() for manual forced token refresh (e.g., after permission changes)
- Add token_obtained_at and token_expires_at tracking fields to Session class
- request_token_client_credentials() now automatically stores expiration timestamps
- Add 9 comprehensive test cases covering all expiration and refresh scenarios
- Default 5-minute buffer before expiration to prevent authentication failures
- Only works with client credentials flow (API versions >= 2026-01)
- Backward compatible: gracefully handles authorization code flow without errors

This enhancement prevents authentication failures in long-running processes by
automatically refreshing tokens before they expire. Tokens from Shopify's client
credentials grant expire after 24 hours (86,399 seconds).

Example usage:
  session = shopify.Session('store.myshopify.com', '2026-01')
  session.request_access_token()

  # Later, before API calls
  result = session.refresh_token_if_needed()
  if result:
      print('Token was refreshed')

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: lvalics

CHANGELOG

@@ -9,6 +9,13 @@
   - Automatic version detection: API versions >= 2026-01 require client credentials flow
   - Legacy request_token() raises ValidationException for API versions >= 2026-01
   - Tokens from client credentials grant expire after 24 hours (86399 seconds)
+- Add token expiration tracking and automatic refresh functionality
+  - New method: Session.is_token_expired() - Check if token is expired or expiring soon (with configurable buffer)
+  - New method: Session.refresh_token_if_needed() - Automatically refresh token if expired or expiring within buffer time
+  - New method: Session.refresh_token() - Manually force token refresh regardless of expiration status
+  - Session now tracks token_obtained_at and token_expires_at timestamps for client credentials tokens
+  - Default 5-minute buffer before expiration to ensure tokens are refreshed proactively
+  - Auto-refresh only works with client credentials flow (API versions >= 2026-01)
 
 == Version 12.7.0
 

shopify/session.py

@@ -2,6 +2,7 @@
 import hmac
 import json
 from hashlib import sha256
+from datetime import datetime, timedelta
 
 try:
     import simplejson as json
@@ -56,6 +57,8 @@ def __init__(self, shop_url, version=None, token=None, access_scopes=None):
         self.token = token
         self.version = ApiVersion.coerce_to_version(version)
         self.access_scopes = access_scopes
+        self.token_expires_at = None
+        self.token_obtained_at = None
         return
 
     def _requires_client_credentials(self):
@@ -215,11 +218,16 @@ def request_token_client_credentials(self):
                 self.token = json_payload["access_token"]
                 self.access_scopes = json_payload.get("scope", "")
 
+                # Store expiration tracking information
+                expires_in = json_payload.get("expires_in", 86399)
+                self.token_obtained_at = datetime.now()
+                self.token_expires_at = self.token_obtained_at + timedelta(seconds=expires_in)
+
                 # Return full response for caller
                 return {
                     "access_token": self.token,
                     "scope": self.access_scopes,
-                    "expires_in": json_payload.get("expires_in", 86399)
+                    "expires_in": expires_in
                 }
             else:
                 raise OAuthException("OAuth request failed with status %d: %s" % (response.code, response.msg))
@@ -288,6 +296,158 @@ def request_access_token(self, params=None):
                 )
             return self.request_token(params)
 
+    def is_token_expired(self, buffer_seconds=300):
+        """
+        Check if access token is expired or will expire soon.
+
+        This method checks whether the current access token has expired or will
+        expire within the specified buffer time. It's useful for proactively
+        refreshing tokens before they expire to avoid authentication failures.
+
+        Args:
+            buffer_seconds (int): Number of seconds before expiration to consider
+                                 the token as "expired". Default is 300 (5 minutes).
+                                 This buffer allows time to refresh the token before
+                                 it actually expires.
+
+        Returns:
+            bool: True if:
+                  - No token is set
+                  - No expiration time is available
+                  - Token has expired
+                  - Token will expire within buffer_seconds
+                  False if token is valid and won't expire soon
+
+        Example:
+            >>> session = shopify.Session("store.myshopify.com", "2026-01")
+            >>> session.request_token_client_credentials()
+            >>> if session.is_token_expired():
+            ...     session.refresh_token()
+            >>> # Or check with custom buffer (10 minutes)
+            >>> if session.is_token_expired(buffer_seconds=600):
+            ...     session.refresh_token()
+        """
+        # No token set - consider expired
+        if not self.token:
+            return True
+
+        # No expiration tracking available - consider expired for safety
+        if not self.token_expires_at:
+            return True
+
+        # Calculate if token is expired or expiring soon
+        now = datetime.now()
+        buffer = timedelta(seconds=buffer_seconds)
+
+        # Token is expired if current time + buffer >= expiration time
+        return now + buffer >= self.token_expires_at
+
+    def refresh_token_if_needed(self, buffer_seconds=300):
+        """
+        Automatically refresh the access token if expired or expiring soon.
+
+        This method checks if the token is expired or will expire within the buffer
+        time, and automatically refreshes it if needed. It's the recommended way to
+        ensure you always have a valid token without manually tracking expiration.
+
+        Args:
+            buffer_seconds (int): Number of seconds before expiration to trigger
+                                 refresh. Default is 300 (5 minutes). This ensures
+                                 the token is refreshed before it expires.
+
+        Returns:
+            dict or None:
+                - dict: Token response if refresh was performed, containing:
+                        - access_token (str): The new access token
+                        - scope (str): Comma-separated list of granted scopes
+                        - expires_in (int): Seconds until token expires
+                - None: If token is still valid and no refresh was needed
+
+        Raises:
+            ValidationException: If required credentials are missing
+            OAuthException: If OAuth refresh request fails
+
+        Example:
+            >>> session = shopify.Session("store.myshopify.com", "2026-01")
+            >>> shopify.Session.setup(api_key="client_id", secret="client_secret")
+            >>>
+            >>> # Initial token request
+            >>> session.request_token_client_credentials()
+            >>>
+            >>> # Later, before making API calls, ensure token is fresh
+            >>> result = session.refresh_token_if_needed()
+            >>> if result:
+            ...     print("Token was refreshed")
+            ... else:
+            ...     print("Token is still valid")
+            >>>
+            >>> # Use custom buffer (refresh if expires within 10 minutes)
+            >>> session.refresh_token_if_needed(buffer_seconds=600)
+        """
+        if self.is_token_expired(buffer_seconds=buffer_seconds):
+            # Only refresh if we have credentials (client credentials flow)
+            if self._requires_client_credentials():
+                return self.request_token_client_credentials()
+            else:
+                # For authorization code flow, we can't auto-refresh
+                # because we need user interaction for the callback
+                return None
+
+        # Token is still valid, no refresh needed
+        return None
+
+    def refresh_token(self):
+        """
+        Manually force a refresh of the access token.
+
+        This method unconditionally refreshes the access token, regardless of
+        whether it has expired or not. Use this when you need to force a token
+        refresh (e.g., after permission changes, for testing, or if you suspect
+        the token is invalid).
+
+        For automatic refresh based on expiration, use refresh_token_if_needed()
+        instead.
+
+        Returns:
+            dict: Token response containing:
+                  - access_token (str): The new access token
+                  - scope (str): Comma-separated list of granted scopes
+                  - expires_in (int): Seconds until token expires (typically 86399)
+
+        Raises:
+            ValidationException: If required credentials are missing or if this
+                               method is called on a session using authorization
+                               code flow (requires user interaction)
+            OAuthException: If OAuth refresh request fails
+
+        Example:
+            >>> session = shopify.Session("store.myshopify.com", "2026-01")
+            >>> shopify.Session.setup(api_key="client_id", secret="client_secret")
+            >>>
+            >>> # Initial token request
+            >>> session.request_token_client_credentials()
+            >>>
+            >>> # Force token refresh (e.g., after permission changes)
+            >>> new_token = session.refresh_token()
+            >>> print(f"New token: {new_token['access_token']}")
+            >>> print(f"Expires in: {new_token['expires_in']} seconds")
+        """
+        # Only works with client credentials flow
+        if not self._requires_client_credentials():
+            raise ValidationException(
+                "Manual token refresh is only supported for API versions >= 2026-01 "
+                "using client credentials flow. For authorization code flow, "
+                "users must re-authorize through the OAuth callback."
+            )
+
+        # Clear existing token to force new request
+        self.token = None
+        self.token_expires_at = None
+        self.token_obtained_at = None
+
+        # Request new token
+        return self.request_token_client_credentials()
+
     @property
     def api_version(self):
         return self.version

test/session_test.py

@@ -471,3 +471,157 @@ def test_request_access_token_uses_authorization_code_for_old_version(self):
         token = session.request_access_token(params)
 
         self.assertEqual(token, "old_style_token")
+
+    def test_is_token_expired_with_no_token(self):
+        """Test that is_token_expired returns True when no token is set"""
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+        self.assertTrue(session.is_token_expired())
+
+    def test_is_token_expired_with_no_expiration(self):
+        """Test that is_token_expired returns True when token has no expiration tracking"""
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+        session.token = "test_token"
+        # token_expires_at is None
+        self.assertTrue(session.is_token_expired())
+
+    def test_is_token_expired_within_buffer(self):
+        """Test that is_token_expired returns True when token expires within buffer"""
+        from datetime import datetime, timedelta
+
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+        session.token = "test_token"
+
+        # Set token to expire in 4 minutes
+        session.token_expires_at = datetime.now() + timedelta(minutes=4)
+
+        # With default 5-minute buffer, should be considered expired
+        self.assertTrue(session.is_token_expired())
+
+        # With 3-minute buffer, should not be considered expired
+        self.assertFalse(session.is_token_expired(buffer_seconds=180))
+
+    def test_is_token_not_expired(self):
+        """Test that is_token_expired returns False when token is valid"""
+        from datetime import datetime, timedelta
+
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+        session.token = "test_token"
+
+        # Set token to expire in 1 hour
+        session.token_expires_at = datetime.now() + timedelta(hours=1)
+
+        # With default 5-minute buffer, should not be expired
+        self.assertFalse(session.is_token_expired())
+
+    def test_refresh_token_if_needed_refreshes_expired_token(self):
+        """Test that refresh_token_if_needed refreshes an expired token"""
+        from datetime import datetime, timedelta
+
+        shopify.Session.setup(api_key="test_client_id", secret="test_client_secret")
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+        session.token = "old_token"
+
+        # Set token as expired
+        session.token_expires_at = datetime.now() - timedelta(minutes=1)
+
+        self.fake(
+            None,
+            url="https://testshop.myshopify.com/admin/oauth/access_token",
+            method="POST",
+            body='{"access_token": "new_token", "scope": "read_products", "expires_in": 86399}',
+            has_user_agent=False,
+        )
+
+        result = session.refresh_token_if_needed()
+
+        # Should return token response
+        self.assertIsNotNone(result)
+        self.assertEqual(result["access_token"], "new_token")
+        self.assertEqual(session.token, "new_token")
+
+    def test_refresh_token_if_needed_does_not_refresh_valid_token(self):
+        """Test that refresh_token_if_needed does not refresh a valid token"""
+        from datetime import datetime, timedelta
+
+        shopify.Session.setup(api_key="test_client_id", secret="test_client_secret")
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+        session.token = "valid_token"
+
+        # Set token to expire in 1 hour
+        session.token_expires_at = datetime.now() + timedelta(hours=1)
+
+        result = session.refresh_token_if_needed()
+
+        # Should return None (no refresh needed)
+        self.assertIsNone(result)
+        self.assertEqual(session.token, "valid_token")
+
+    def test_refresh_token_forces_refresh(self):
+        """Test that refresh_token forces a token refresh"""
+        from datetime import datetime, timedelta
+
+        shopify.Session.setup(api_key="test_client_id", secret="test_client_secret")
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+        session.token = "old_valid_token"
+
+        # Set token to expire in 1 hour (still valid)
+        session.token_expires_at = datetime.now() + timedelta(hours=1)
+
+        self.fake(
+            None,
+            url="https://testshop.myshopify.com/admin/oauth/access_token",
+            method="POST",
+            body='{"access_token": "forced_new_token", "scope": "read_products", "expires_in": 86399}',
+            has_user_agent=False,
+        )
+
+        result = session.refresh_token()
+
+        # Should force refresh even though token was valid
+        self.assertEqual(result["access_token"], "forced_new_token")
+        self.assertEqual(session.token, "forced_new_token")
+
+    def test_refresh_token_fails_for_old_api_version(self):
+        """Test that refresh_token raises error for API versions < 2026-01"""
+        shopify.Session.setup(api_key="test_key", secret="test_secret")
+        session = shopify.Session("testshop.myshopify.com", "2025-10")
+        session.token = "old_version_token"
+
+        with self.assertRaises(shopify.ValidationException) as context:
+            session.refresh_token()
+
+        self.assertIn("2026-01", str(context.exception))
+        self.assertIn("client credentials flow", str(context.exception))
+
+    def test_token_expiration_tracking_stored(self):
+        """Test that token expiration tracking is properly stored"""
+        from datetime import datetime, timedelta
+
+        shopify.Session.setup(api_key="test_client_id", secret="test_client_secret")
+        session = shopify.Session("testshop.myshopify.com", "2026-01")
+
+        before_request = datetime.now()
+
+        self.fake(
+            None,
+            url="https://testshop.myshopify.com/admin/oauth/access_token",
+            method="POST",
+            body='{"access_token": "test_token", "scope": "read_products", "expires_in": 86399}',
+            has_user_agent=False,
+        )
+
+        session.request_token_client_credentials()
+
+        after_request = datetime.now()
+
+        # Verify expiration tracking is set
+        self.assertIsNotNone(session.token_obtained_at)
+        self.assertIsNotNone(session.token_expires_at)
+
+        # Verify obtained_at is between before and after
+        self.assertGreaterEqual(session.token_obtained_at, before_request)
+        self.assertLessEqual(session.token_obtained_at, after_request)
+
+        # Verify expires_at is approximately 24 hours from obtained_at
+        expected_expiration = session.token_obtained_at + timedelta(seconds=86399)
+        self.assertEqual(session.token_expires_at, expected_expiration)