Testing SQL Connections
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 presentThis 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)