We want to test database connections in all our instances since our internal security policy is to change database passwords yearly and this invariably leads to some user connections being missed and this invariably leads to some user database connections being missed, the password expiring, flows failing, data not being updated, etc. So we are trying to be more proactive and try to test connections programatically. I looked at the REST API and there isn't any available public APIs at the moment and I have raised the Product Idea here so please vote for it if you like it.
As DSS has this feature built in in the GUI I did some digging and noticed the GUI calls the /dip/api/admin/connections/test-sql internal/private API when I click on the Test button of a SQL connection. I am now trying to see if I can get this internal/private API working. I obviously understand internal/private API are unsupported and could change/break/dissapear at any point but I see no harm in trying.
I am stuck as I keep getting "500 - Server Error". Looking at the backend log I can see what the issue is:
[2023/05/31-14:13:24.295] [qtp1900366749-124139] [ERROR] [dip.controllers] - API call '/dip/api/admin/connections/test-sql' failedorg.springframework.web.bind.MissingServletRequestParameterException: Required String parameter 'data' is not present
This is happening since the API expects all the connection details to be sent into a data URL encoded parameter but I have not been able to replicate this. My code is below, anyone knows how to get this working? Thanks!
import dataikuapi
import requests
import json
requests.packages.urllib3.disable_warnings()
api_key = ''
base_url = 'https://dss_url/'
test_connection_url = base_url + 'dip/api/admin/connections/test-sql'
headers = {'Accept': 'application/json, text/plain, */*',
'Accept-Encoding': 'gzip, deflate, br',
'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8',
'Authorization': 'Bearer {}'.format(api_key)}
client = dataikuapi.DSSClient(base_url, api_key)
client._session.verify = False
connections = client.list_connections()
for connection in connections:
if connection == 'pick_one_sql_connection':
try:
payload_str = urllib.parse.urlencode(connections[connection], safe=':+')
response = requests.post(test_connection_url, headers=headers, data=payload_str, verify=False, timeout=(1, 3))
status_code = response.status_code
status_reason = response.reason
status_text = str(status_code) + ' - ' + str(status_reason)
# Raise an exception if the response status code is not successful
response.raise_for_status()
except requests.exceptions.BaseHTTPError:
error_text = "Base HTTP Error: " + status_text
except requests.exceptions.HTTPError:
error_text = "HTTP Error: " + status_text
except requests.exceptions.Timeout:
error_text = "The request timed out"
except requests.exceptions.ConnectionError:
error_text = "Connection Error"
except requests.exceptions.RequestException:
error_text = "Unknown error occurred: " + str(status_code)
# If the request was successful
if status_code == 200:
# Parse the response as JSON
json_object = response.json()
# Print JSON object
print(json.dumps(json_object, indent=4))
else:
print(status_reason)
print(error_text)
