blob: 0b3e4a3971028c663ea0f699283a9fc5ed89449a (
plain)
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
|
import base64
import datetime
import os
from .shared_flow import send_token_request
def get_access_token_from_refresh_token(refresh_token: str) -> dict[str, any]:
"""
Obtains a new access token using a refresh token.
This function uses the provided refresh token to request a new access token
from the authentication server. It updates the expiration time of the token.
Args:
refresh_token: The refresh token used to obtain a new access token.
Returns:
A dictionary containing the new access token data, including 'expires_at'.
"""
client_id = os.getenv('CLIENT_ID')
secret_key = os.getenv('SECRET_KEY')
if not client_id or not secret_key:
raise EnvironmentError("CLIENT_ID and SECRET_KEY must be set in environment variables.")
user_pass = f"{client_id}:{secret_key}"
basic_auth = base64.urlsafe_b64encode(user_pass.encode('utf-8')).decode()
auth_header = f"Basic {basic_auth}"
form_values = {
"grant_type": "refresh_token",
"refresh_token": refresh_token,
}
headers = {"Authorization": auth_header}
response = send_token_request(form_values, add_headers=headers)
data = response.json()
current_timestamp = int(datetime.datetime.now().timestamp())
data["expires_at"] = current_timestamp + data.get("expires_in", 0)
return data
|
