GET /api/v2/users/me returns 401 Unauthorized after processing 150 records.
I am using a simple Python loop with requests.get() and a static Bearer token. The token expires exactly when the script hits the next iteration. I tried calling POST /api/v2/oauth/token with grant_type=refresh_token but get invalid_grant.
How do I correctly implement the refresh logic in Python without losing the batch state?
This looks like a token refresh race condition in your loop. You cannot rely on a static Bearer token for batch processing. The invalid_grant error usually means the refresh token was already consumed or expired, often because the initial auth flow didn’t persist the new refresh token returned in the response.
Use the PureCloudPlatformClientV2 SDK. It handles token rotation automatically. If you must use raw requests, you need to catch the 401, refresh the token, and retry the failed request. Do not just loop blindly.
import requests
from functools import wraps
def token_refresh_session(client_id, client_secret, refresh_token):
"""Session object that auto-refreshes token on 401."""
session = requests.Session()
def refresh_token_logic():
# Exchange refresh_token for new access_token
resp = session.post('https://api.mypurecloud.com/api/v2/oauth/token',
data={
'grant_type': 'refresh_token',
'refresh_token': refresh_token,
'client_id': client_id,
'client_secret': client_secret
})
resp.raise_for_status()
data = resp.json()
session.headers['Authorization'] = f"Bearer {data['access_token']}"
# CRITICAL: Update the stored refresh token if a new one is provided
if 'refresh_token' in data:
nonlocal refresh_token
refresh_token = data['refresh_token']
return data['access_token']
# Initial token fetch omitted for brevity
# Assume session.headers['Authorization'] is set
def request_hook(response, *args, **kwargs):
if response.status_code == 401:
refresh_token_logic()
return session.request(response.request.method, response.request.url, **response.request.kwargs)
return response
session.hooks['response'].append(request_hook)
return session
# Usage
sess = token_refresh_session(CLIENT_ID, CLIENT_SECRET, REFRESH_TOKEN)
for i in range(1000):
resp = sess.get('https://api.mypurecloud.com/api/v2/users/me')
# Process data
The key is updating the refresh_token variable from the response. Many clients ignore this, causing subsequent refreshes to fail. In Django, store this in Redis or a secure DB field, not in memory only, so Celery workers can persist across restarts. Also, ensure your OAuth client has the refresh_token grant type enabled in the Developer Portal.