50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
from posting import Auth, Header, RequestModel, Posting
|
|
import httpx
|
|
|
|
def get_oauth2_token(client_id, client_secret, token_url, audience):
|
|
"""
|
|
Request an OAuth2 access token using the Client Credentials Grant flow.
|
|
|
|
Args:
|
|
client_id (str): The client ID provided by the OAuth2 provider.
|
|
client_secret (str): The client secret provided by the OAuth2 provider.
|
|
token_url (str): The URL to request the access token.
|
|
|
|
Returns:
|
|
str: The access token if the request is successful.
|
|
"""
|
|
# Prepare the data for the token request
|
|
data = {
|
|
"grant_type": "client_credentials", # Grant type for Client Credentials flow
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
"audience": audience
|
|
}
|
|
|
|
try:
|
|
# Make the POST request to the token endpoint
|
|
response = httpx.post(token_url, data=data)
|
|
|
|
# Raise an exception if the request failed
|
|
response.raise_for_status()
|
|
|
|
# Parse the JSON response
|
|
token_data = response.json()
|
|
|
|
print ("Successfully obtained access_token")
|
|
|
|
# Return the access token
|
|
return token_data.get("access_token")
|
|
except httpx.RequestError as e:
|
|
print(f"An error occurred while requesting the token: {e}")
|
|
except httpx.HTTPStatusError as e:
|
|
print(f"HTTP error occurred: {e.response.status_code} - {e.response.text}")
|
|
except KeyError:
|
|
print("The response did not contain an access token.")
|
|
return None
|
|
|
|
def setup(posting: Posting) -> None:
|
|
if not posting.get_variable("auth_token"):
|
|
token = get_oauth2_token(posting.variables["CLIENT_ID_V2"], posting.variables["CLIENT_SECRET_V2"],posting.variables["TOKEN_URL"],posting.variables["AUDIENCE"],)
|
|
posting.set_variable("auth_token", token)
|