update client and oath and improve code

This commit is contained in:
Greg Gauthier 2024-07-19 23:45:11 +01:00
parent a282312e78
commit a858809f06
5 changed files with 105 additions and 47 deletions

35
api-client/client.py Normal file
View File

@ -0,0 +1,35 @@
import sys
from requests import RequestException
from requests_oauthlib import OAuth2Session
def api_client(call_dict):
url = call_dict["url"]
headers = call_dict["headers"]
body = call_dict["body"]
client = OAuth2Session(token=call_dict["token"])
method = call_dict["method"]
try:
if method == 'GET':
response = client.get(url, headers=headers, params=body)
elif method == 'POST':
response = client.post(url, headers=headers, json=body)
elif method == 'PUT':
response = client.put(url, headers=headers, json=body)
elif method == 'OPTIONS':
response = client.options(url, headers=headers, json=body)
elif method == 'DELETE':
response = client.delete(url)
else:
raise ValueError(f"Invalid method: {method}")
except RequestException as e:
print(f"Request failed. Method: {method}, URL: {url}", file=sys.stderr)
print(f"Error details: {str(e)}", file=sys.stderr)
exit(e.response.status_code)
return response.content, response.status_code

22
api-client/config.py Normal file
View File

@ -0,0 +1,22 @@
import os
from dotenv import load_dotenv
from types import MappingProxyType
def get_cfg(env='qa'):
environment_name = os.getenv('ENV_NAME', env)
dotenv_path = f'.env.{environment_name}'
load_dotenv(dotenv_path)
config_dict = {
'client_id': os.getenv('CLIENT_ID'),
'client_secret': os.getenv('CLIENT_SECRET'),
'token_url': os.getenv('TOKEN_FETCH_URL'),
'username': os.getenv('USERNAME'),
'password': os.getenv('PASSWORD'),
'audience': os.getenv('AUDIENCE'),
'scopes': os.getenv('SCOPES', '').split(','),
'api_url': os.getenv('API_URL')
}
config = MappingProxyType(config_dict) # immutable dict
return config

View File

@ -1,47 +0,0 @@
from requests_oauthlib import OAuth2Session
from oauthlib.oauth2 import BackendApplicationClient
from dotenv import load_dotenv
import os
environment_name = os.getenv('ENV_NAME', 'qa') # need to find a place for this
dotenv_path = f'.env.{environment_name}'
load_dotenv(dotenv_path)
client_id = os.getenv('CLIENT_ID')
client_secret = os.getenv('CLIENT_SECRET')
token_url = os.getenv('TOKEN_URL')
username = os.getenv('USERNAME')
password = os.getenv('PASSWORD')
audience = os.getenv('AUDIENCE')
scope = os.getenv('SCOPE', '').split(',')
api_url = os.getenv('API_URL')
def get_token(client_id, client_secret, token_url, username, password, audience, scope):
client = BackendApplicationClient(client_id=client_id)
oauth = OAuth2Session(client=client)
token = oauth.fetch_token(
token_url=token_url,
username=username,
password=password,
client_id=client_id,
client_secret=client_secret,
audience=audience,
scope=scope
)
return token
def api_client(resource_url, auth_token):
client = OAuth2Session(token=auth_token)
response = client.get(resource_url)
return response.content
# Fetch token
token = get_token(client_id, client_secret, token_url, username, password, audience, scope)
# Now we can make API calls
api_response = api_client(api_url + '/resource', token)
print(api_response)

21
api-client/main.py Normal file
View File

@ -0,0 +1,21 @@
from config import get_cfg
from client import api_client
from oauth_helper import get_legacy_token
ENV = 'qa'
cfg = get_cfg(ENV)
if __name__ == "__main__":
token = get_legacy_token(ENV)
api_call = {
"token": token,
"method": "GET",
"url": cfg["api_url"] + '/data-delivery/sarredaction/119',
"headers": {'Content-Type': 'application/json'},
"body": {}
}
api_response = api_client(api_call)
print(api_response)

View File

@ -0,0 +1,27 @@
from oauthlib.oauth2 import LegacyApplicationClient, OAuth2Error
from requests_oauthlib import OAuth2Session
from config import get_cfg
import sys
def get_legacy_token(env):
cfg = get_cfg(env)
client = LegacyApplicationClient(client_id=cfg['client_id'])
oauth = OAuth2Session(client=client)
try:
token = oauth.fetch_token(
token_url=cfg['token_url'],
username=cfg['username'],
password=cfg['password'],
client_id=cfg['client_id'],
client_secret=cfg['client_secret'],
audience=cfg['audience'],
scope=cfg['scopes']
)
except OAuth2Error as e:
print("OAuth2 Error: ", str(e), file=sys.stderr)
exit(e.status_code)
return token