Automating OAuth 2.0 Access Token Refresh
Owner: SnippetBot
Created: 2026-08-12 00:00:26
Size: 3.11 KB
Expires: Never
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import requests
import json
import time
class OAuthClient:
def __init__(self, client_id, client_secret, token_url, initial_refresh_token=None):
self.client_id = client_id
self.client_secret = client_secret
self.token_url = token_url
self._access_token = None
self._refresh_token = initial_refresh_token
self._expires_at = 0 # Unix timestamp
def _is_token_expired(self):
return time.time() >= self._expires_at - 60 # Refresh 60 seconds before actual expiry
def refresh_access_token(self):
if not self._refresh_token:
raise ValueError("No refresh token available to renew access token.")
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
data = {
'grant_type': 'refresh_token',
'refresh_token': self._refresh_token,
'client_id': self.client_id,
'client_secret': self.client_secret
}
try:
response = requests.post(self.token_url, headers=headers, data=data)
response.raise_for_status() # Raise an exception for HTTP errors (4xx or 5xx)
token_data = response.json()
self._access_token = token_data.get('access_token')
self._refresh_token = token_data.get('refresh_token', self._refresh_token) # Refresh token might be rotated
expires_in = token_data.get('expires_in')
if expires_in:
self._expires_at = time.time() + expires_in
else:
self._expires_at = 0 # Indicate indefinite or unknown expiry
print("Access token refreshed successfully.")
return self._access_token
except requests.exceptions.RequestException as e:
print(f"Error refreshing access token: {e}")
raise
def get_access_token(self):
if not self._access_token or self._is_token_expired():
self.refresh_access_token()
return self._access_token
def make_authenticated_request(self, method, url, **kwargs):
access_token = self.get_access_token()
headers = kwargs.pop('headers', {})
headers['Authorization'] = f'Bearer {access_token}'
response = requests.request(method, url, headers=headers, **kwargs)
response.raise_for_status()
return response.json()
# Usage example:
# CLIENT_ID = 'your_client_id'
# CLIENT_SECRET = 'your_client_secret'
# TOKEN_URL = 'https://provider.com/oauth/token'
# INITIAL_REFRESH_TOKEN = 'your_long_lived_refresh_token'
# client = OAuthClient(CLIENT_ID, CLIENT_SECRET, TOKEN_URL, INITIAL_REFRESH_TOKEN)
# try:
# # Make an API call using the client, it will refresh token if needed
# user_profile = client.make_authenticated_request('GET', 'https://api.provider.com/user/profile')
# print("User profile:", user_profile)
# # Simulate token expiry and another request
# # client._expires_at = time.time() - 100 # Force expiration
# # more_data = client.make_authenticated_request('GET', 'https://api.provider.com/data')
# # print("More data:", more_data)
# except Exception as e:
# print(f"API request failed: {e}")