-
Notifications
You must be signed in to change notification settings - Fork 140
Open
Description
Add OAuth identification as per the opensky new authentication strategy.
I added the following snippets
1. New method of OpenSkyApi class :
`def _get_opensky_token(self, client_id:str, client_secret:str):
"""
Retrieves an access token from the OpenSky Network API.
This function sends a POST request to the OpenSky Network's token endpoint
to obtain a bearer token for API access.
Returns:
str: The access token if the request is successful, otherwise None.
"""
if not client_id or not client_secret:
print("Error: OPENSKY_CLIENT_ID and OPENSKY_CLIENT_SECRET environment variables must be set.")
return None
# The URL for the token endpoint
token_url = "https://auth.opensky-network.org/auth/realms/opensky-network/protocol/openid-connect/token"
# The data payload for the POST request
payload = {
"grant_type": "client_credentials",
"client_id": client_id,
"client_secret": client_secret,
}
# The headers for the request
headers = {
"Content-Type": "application/x-www-form-urlencoded"
}
try:
# Make the POST request
response = requests.post(token_url, headers=headers, data=payload)
# Raise an exception for bad status codes (4xx or 5xx)
response.raise_for_status()
# Parse the JSON response and extract the access token
access_token = response.json().get("access_token")
return access_token
except requests.exceptions.RequestException as e:
print(f"An error occurred during the request: {e}")
# If the response has content, print it for more details
if e.response:
print(f"Response content: {e.response.text}")
return None
except ValueError:
# Catches JSON decoding errors
print("Error: Failed to decode JSON response.")
print(f"Response content: {response.text}")
return None`
2. Changes in __init__ constructor:
Add the following code at the end of the __init__ method:
self._headers = { "Authorization": f"Bearer {self._get_opensky_token(client_id=username, client_secret=password)}" }
3. Change in _get_json method:
Change the line auth=self._auth with headers=self._headers.
Metadata
Metadata
Assignees
Labels
No labels