Examples of using Python scripts to make DPA API calls

The following examples show Python scripts that call the DPA API to retrieve information and perform DPA management functions. The first examples are snippets that demonstrate each API call individually. The last example is a full script that shows how to put the snippets together into a working script.

You can call the DPA API with any programming language that can send HTTP requests. See this topic for PowerShell script examples.

See the following sections:

Prerequisites

• Before you can use scripts to make API calls, you must create a refresh token.
• These examples use the Requests HTTP library for Python. This library must be installed for these examples to work.

If your DPA server does not use HTTPS or your certificates are self-signed

The examples all use HTTPS, which can cause problems if your DPA server is not configured to use HTTPS or if your certificates are self signed. If this is the case, you can do either of the following:

• Run the examples using HTTP.
• Change the verify_cert value to False in the configuration section to prevent verifying the server's TLS certificate.

# ==========================================================
# Configure the variables below for the DPA Host
# ==========================================================
base_url = "https://localhost:8124/iwc/api/"
refresh_token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJNeVRva2VuIiwiaXN..."
verify_cert = False
# ==========================================================

Get an access token

The first step in using the API is to get an access token. An access token is required to make any API calls. This call POSTs the refresh token to DPA, which returns an access token to be used by all other API calls.

• If the call is successful, it prints out the data that was returned from DPA, including the access_token, and then goes on to create HTTP Headers that will contain the access token and other information to be used on subsequent calls.

• If the call is not successful it prints out the error message.

You must set the base_url and the refresh_token variables to match your environment.

# ==========================================================
# Configure the variables below for the DPA Host
# ==========================================================
base_url = "https://localhost:8124/iwc/api/"
refresh_token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJNeVRva2VuIiwiaXN..."
verify_cert = True
# ==========================================================

# ==========================================================
# Get Access Token
# ==========================================================
def get_access_header(prefix_url, rfrsh_token):
  """
  Given a base url and a refresh token retrieve the access token
  and return a header object with it.
  :param prefix_url: the base url
  :param rfrsh_token: refresh token used to get access token
  :return: the request header that contains the access token
  :rtype: dict
  """

  auth_token_url = prefix_url + "security/oauth/token"
  grant_type = "refresh_token"

  payload = {"grant_type": grant_type, "refresh_token": rfrsh_token}
  try:
    # get an access token
    resp = requests.post(auth_token_url, data=payload, verify=verify_cert)
    resp.raise_for_status()
    resp_json = resp.json()

    token_type = resp_json["token_type"]
    access_code = resp_json["access_token"]

    headers = {"authorization": f"{token_type} {access_code}",
      "content-type": "application/json;charset=UTF-8",
      "accept": "application/json"
      }

    return headers

  except requests.exceptions.HTTPError as ex:
    print(ex)
    print(ex.response.text)
    # print(json.dumps(json.loads(ex.response.text), indent=2))
    return None  # requests is bad return None, can't get access_code

# get the header that contains access token for authentication
header = get_access_header(base_url, refresh_token)
if header is None:
  sys.exit(0)

Database Monitor examples

The following examples show how to use Database Monitor calls.

Get information about one monitored database instance

# Get information about a single monitored database instance
database_id = 1
monitor_url = f"{base_url}databases/{database_id}/monitor-information"
single_monitor = None
try:
  print(f"\n*** Get Monitor Information for database with id of {database_id} ***")
  response = requests.get(monitor_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  single_monitor = response_json["data"]
  print(json.dumps(single_monitor, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Get Monitor Information for database with id of 1 ***
{
  "dbId": 1,
  "name": "DEV-DPA\SQLEXPRESS",
  "ip": "127.0.0.1",
  "port": "1433",
  "jdbcUrlProperties": "applicationIntent=readOnly",
  "connectionProperties": null,
  "databaseType": "SQL Server",
  "databaseVersion": "12.0.6205.1",
  "databaseEdition": "Enterprise Edition: Core-based Licensing (64-bit)",
  "monitoringUser": "ignite_next",
  "defaultDbLicenseCategory": "DPACAT2",
  "assignedDbLicenseCategory": "DPACAT2",
  "assignedVmLicenseCategory": null,
  "monitorState": "Monitor Stopped",
  "oldestMonitoringDate": "2018-12-09T00:00:00.000-07:00",
  "latestMonitoringDate": "2019-01-07T00:00:00.000-07:00",
  "agListenerName": null,
  "agClusterName": null,
  "agName": null,
  "racInfo": null,
  "rac": false,
  "rds": false,
  "ebusiness": false,
  "linkedToVirtualMachine": false,
  "pdb": false
}

Start and stop monitoring a database instance given its database ID

database_id = 1
monitor_url = f"{base_url}databases/{database_id}/monitor-status"
try:
  # Start monitoring a database instance given its database ID.
  print(f"*** Start Monitor for database {database_id} ***")
  body = {"command": "START"}
  response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))

  print("Waiting 15 seconds...")
  time.sleep(15)

  # Stop monitoring a database instance given its database ID.
  print(f"*** Stop Monitor for database {database_id} ***")
  body = {"command": "STOP"}
  response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
  print("Waiting 15 seconds...")
  time.sleep(15)

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Start Monitor for database 1 ***
"SUCCESS"
Waiting 15 seconds...

*** Stop Monitor for database 1 ***
"SUCCESS"
Waiting 15 seconds...

Get information about all monitored database instances

database_id = 1
monitor_url = f"{base_url}databases/monitor-information"
try:
  print("*** Get Information for a all database instances ***")
  response = requests.get(monitor_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  data = response_json["data"]
  print(json.dumps(data, indent=2))

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Get information for all database instances ***
[
  {
    "dbId": 1,
    "name": "DEV-DPA\SQLEXPRESS",
    "ip": "127.0.0.1",
    "port": "1433",
    "jdbcUrlProperties": "applicationIntent=readOnly",
    "connectionProperties": null,
    "databaseType": "SQL Server",
    "databaseVersion": "12.0.6205.1",
    "databaseEdition": "Enterprise Edition: Core-based Licensing (64-bit)",
    "monitoringUser": "ignite_next",
    "defaultDbLicenseCategory": "DPACAT2",
    "assignedDbLicenseCategory": "DPACAT2",
    "assignedVmLicenseCategory": null,
    "monitorState": "Monitor Stopped",
    "oldestMonitoringDate": "2018-12-09T00:00:00.000-07:00",
    "latestMonitoringDate": "2019-01-07T00:00:00.000-07:00",
    "agListenerName": null,
    "agClusterName": null,
    "agName": null,
    "racInfo": null,
    "rac": false,
    "rds": false,
    "ebusiness": false,
    "linkedToVirtualMachine": false,
    "pdb": false
  },
  {
    "dbId": 2,
    "name": "DEV-MYSQL",
    "ip": "127.0.0.1",
    ...
  }
]

Stop and start monitoring for all database instances

# Start monitoring all database instances.
monitor_url = f"{base_url}databases/monitor-status"
try:
  print("*** Starting all Monitors ***")
  body = {"command": "START"}
  response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
  print("Waiting 30 seconds...")
  time.sleep(30)

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Stop monitoring all database instances.
try:
  print("*** Stopping all Monitors ***")
  body = {"command": "STOP"}
  response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
  print("Waiting 30 seconds...")
  time.sleep(30)

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Starting all Monitors ***
"SUCCESS"
Waiting 30 seconds...

*** Stopping all Monitors ***
"SUCCESS"
Waiting 30 seconds...

Update the user password for a monitored database instance

database_id = 1
monitor_url = f"{base_url}databases/{database_id}/update-password"
try:
  print(f"*** Update the Monitor password for database {database_id} ***")
  body = {"password": "NewPassword!"}
  response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Update the Monitor password for database 1 ***
"SUCCESS"

License Allocation examples

The examples below show how to use License Allocation calls.

Get information about currently installed licenses

license_url = f"{base_url}databases/licenses/installed"
try:
  print("\n*** Getting Installed license information with total amounts available for use and total amounts used ***")
  response = requests.get(license_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  data = response_json["data"]
  print("licenseProduct licenseCategory licensesAvailable licensesConsumed")
  print("-------------- --------------- ----------------- ----------------")
  for i in range(len(data)):
    print('{:<15s}{:<16s}{:>17d}{:>17d}'.format(data[i]["licenseProduct"], data[i] ["licenseCategory"],
				                                            data[i]["licensesAvailable"], data[i]["licensesConsumed"]))

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Getting Installed license information with total amounts available for use and total amounts used ***
licenseProduct licenseCategory licensesAvailable licensesConsumed
-------------- --------------- ----------------- ----------------
DPACAT1        DPA_DB                        100               22
DPACAT2        DPA_DB                        100               16
DPAAzureSQL    DPA_DB                          0                0
DPAVM          DPA_VM                        100               12

Get license information for a single database instance

database_id = 1
license_url = f"{base_url}databases/{database_id}/licenses"
try:
  print(f"\n*** Getting current license information for the database instance with database ID of {database_id} ***")
  response = requests.get(license_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Getting current license information for the database instance with database ID of 1 ***
{
  "serverName": "DEV-DPA",
  "overLicensed": false,
  "performanceLicenseProduct": "DPACAT2",
  "vmLicenseProduct": "DPAVM"
}

Update license information for a database instance

database_id = 1
license_url = f"{base_url}databases/{database_id}/licenses"

# Add a DPACAT2 and a DPAVM license
body = {"performanceLicenseProduct": "DPACAT2",
	"vmLicenseProduct": "DPAVM"}
try:
  print(f"\n*** Updating license for database id {database_id} ***")
  response = requests.put(license_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Remove the DPAVM license
body = {"performanceLicenseProduct": "DPACAT2",
	 "vmLicenseProduct": "REMOVE"}
try:
  print(f"\n*** Updating license for database id {database_id} ***")
  response = requests.put(license_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Updating license for database id 1 ***
{
  "serverName": "DEV-BOU-CALLEN",
  "overLicensed": false,
  "performanceLicenseProduct": "DPACAT2",
  "vmLicenseProduct": DPAVM
}
*** Updating license for database id 1 ***
{
  "serverName": "DEV-BOU-CALLEN",
  "overLicensed": false,
  "performanceLicenseProduct": "DPACAT2",
  "vmLicenseProduct": null
}

Annotation examples

The examples below show how to use Annotation calls.

Get a list of annotations for the last 30 days

# Gets a List of annotations for the last 30 days
database_id = 1
annotation_url = f"{base_url}databases/{database_id}/annotations"

# Dates are in ISO 8601 format ( 2018-12-31T12:00:00.000-07:00 )
end_time = datetime.datetime.now()
start_time = end_time + datetime.timedelta(days=-30)
args = {"startTime": start_time.astimezone().isoformat(),
	 "endTime": end_time.astimezone().isoformat()}

try:
  print("\n*** Getting Annotations for the last 30 days ***")
  response = requests.get(annotation_url, params=args, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Getting Annotations for the last 30 days ***
[
  {
    "id": 112,
    "title": "Test Title API",
    "description": "Test Event created by DPA API",
    "createdBy": "DPA API",
    "time": "2018-12-11T10:01:35-07:00",
    "type": "API"
  },
  {
    "id": 113,
    "title": "Test Title API",
    "description": "Test Event created by DPA API",
    "createdBy": "DPA API",
    "time": "2018-12-12T15:00:40-07:00",
    "type": "API"
  },
  {
    ...
  }
]

Create a new annotation

database_id = 1
annotation_url = f"{base_url}databases/{database_id}/annotations"

# Dates are in ISO 8601 format, no millis ( 2018-12-31T12:00:00-07:00 )
create_time = datetime.datetime.now().replace(microsecond=0)
body = {"title": "API Test Title",
  "description": "API Test Description",
  "createdBy": "Test API User",
  "time": create_time.astimezone().isoformat()}
try:
  print("\n*** Creating Annotation ***")
  response = requests.post(annotation_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Creating Annotation ***
{
  "id": 171,
  "title": "API Test Title",
  "description": "API Test Description",
  "createdBy": "Test API User",
  "time": "2019-01-09T11:04:33-07:00",
  "type": "API"
}

Delete an annotation

database_id = 1
annotation_id = 171
annotation_url = f"{base_url}databases/{database_id}/annotations/{annotation_id}"
try:
  print(f"\n*** Deleting Annotation with id of {annotation_id} ***")
  response = requests.delete(annotation_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  if response.status_code == 204:
    print(f"Annotation with id of {annotation_id} deleted")
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Deleting Annotation with id of 171 ***
Annotation with id of 171 deleted

Database Registration examples

The examples below show how to use Database Registration calls.

Register and unregister a SQL Server database instance for monitoring

This example registers a new SQL Server database instance, waits 60 seconds, and then unregisters the database instance.

# ----------------------------------------------------------
# Register a SQL Server database instance for monitoring.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/register-monitor"
body = {"databaseType": "SQLSERVER",
        "serverName": "127.0.0.1",
        "port": "1433",
        "sysAdminUser": "sa",
        "sysAdminPassword": "Password",
        "monitoringUser": "dpa_test_m",
        "monitoringUserPassword": "Password",
        "monitoringUserIsNew": True,
        "displayName": "DPA_SQL2K12"}

new_db_id = None
try:
  print("\n*** Register SQL Server database ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  data = responseJson["data"]
  new_db_id = data["databaseId"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

print("Waiting 60 seconds...")
time.sleep(60)

# ----------------------------------------------------------
# Unregister the SQL Server database instance.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/unregister-monitor"
body = {"databaseId": new_db_id,
        "removeMonitoringUser": True,
        "removeDatabaseObjects": True,
        "sysAdminUser": "sa",
        "sysAdminPassword": "Password"}
try:
  print(f"\n*** Unregister SQL Server database [{new_db_id}] ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  print(json.dumps(responseJson["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Register SQL Server database ***
{
  "databaseId": 77,
  "result": "SUCCESS"
}
Waiting 60 seconds...

*** Unregister SQL Server database [77] ***
{
  "databaseId": 77,
  "result": "SUCCESS"
}

Register and unregister an Oracle database instance for monitoring

This example registers a new Oracle database instance, waits 60 seconds, and then unregisters the database instance.

# ----------------------------------------------------------
# Register an Oracle database instance for monitoring.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/register-monitor"
body = {"databaseType": "ORACLE",
        "serverName": "127.0.0.1",
        "serviceNameOrSID": "DPA_ORA11R1",
        "port": "1521",
        "sysAdminUser": "system",
        "sysAdminPassword": "Password",
        "sysPassword": "Password",
        "monitoringUser": "dpa_test_m",
        "monitoringUserPassword": "Password",
        "monitoringUserIsNew": True,
        "monitoringUserTableSpace": "USERS",
        "monitoringUserTempTableSpace": "TEMP",
        "oracleEBusinessEnabled": False,
        "displayName": "DPA_ORA11R1"}

new_db_id = None
try:
  print("\n*** Register Oracle database ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  data = responseJson["data"]
  new_db_id = data["databaseId"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

print("Waiting 60 seconds...")
time.sleep(60)

# ----------------------------------------------------------
# Unregister the Oracle database instance.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/unregister-monitor"
body = {"databaseId": new_db_id,
        "removeMonitoringUser": True,
        "removeDatabaseObjects": True,
        "sysAdminUser": "system",
        "sysAdminPassword": "Password"}
try:
  print(f"\n*** Unregister Oracle database [{new_db_id}] ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  print(json.dumps(responseJson["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Register Oracle database ***
{
  "databaseId": 78,
  "result": "SUCCESS"
}
Waiting 60 seconds...

*** Unregister Oracle database [78] ***
{
  "databaseId": 78,
  "result": "SUCCESS"
}

Register and unregister a MySQL database instance for monitoring

This example registers a new MySQL database instance, waits 60 seconds, and then unregisters the database instance.

# ----------------------------------------------------------
# Register a MySQL database instance for monitoring.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/register-monitor"
body = {"databaseType": "MYSQL",
        "serverName": "127.0.0.1",
        "port": "3306",
        "sysAdminUser": "root",
        "sysAdminPassword": "Password",
        "monitoringUser": "dpa_test_m",
        "monitoringUserPassword": "Password",
        "monitoringUserIsNew": True,
        "displayName": "DPA_MYSQL56"}

new_db_id = None
try:
  print("\n*** Register MySQL database ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  data = responseJson["data"]
  new_db_id = data["databaseId"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

print("Waiting 60 seconds...")
time.sleep(60)

# ----------------------------------------------------------
# Unregister the MySQL database instance.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/unregister-monitor"
body = {"databaseId": new_db_id,
        "removeMonitoringUser": True,
        "removeDatabaseObjects": True,
        "sysAdminUser": "root",
        "sysAdminPassword": "Password"}
try:
  print(f"\n*** Unregister MySQL database [{new_db_id}] ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  print(json.dumps(responseJson["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will print out data like this:
*** Register MySQL database ***
{
  "databaseId": 79,
  "result": "SUCCESS"
}
Waiting 60 seconds...

*** Unregister MySQL database [79] ***
{
  "databaseId": 79,
  "result": "SUCCESS"
}

Database Custom Properties examples

The examples below show how to use Database Custom Properties calls. Custom property values can be included in custom email templates for alert notifications.

Create a custom property

This script creates a custom property and defines its name and description.

property_name = "Location"
property_description = "Location of the database server"
create_property_url = f"{base_url}databases/properties"
body = {
  "name": property_name,
  "description": property_description
}

property_id = None

try:
  print("\n*** Creating custom property ***")
  response = requests.post(create_property_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  data = response_json["data"]
  property_id = data["id"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

#This will print out data like this:
{
  "id": 1,
  "name": "Location",
  "description": "Location of the database server"
}

Create a custom property value

This script creates a value for the custom property created by the previous script.

property_id = 1
property_value = "New York"
create_value_url = f"{base_url}databases/properties/" + str(property_id) + "/values"
body = property_value

property_value_id = None

try:
  print("\n*** Creating custom property value ***")
  response = requests.post(create_value_url, data=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  data = response_json["data"]
  property_value_id = data["id"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

#This will print out data like this:
{
  "id": 1,
  "value": "New York"
}

Assign a property value to a monitored database instance

This script assigns a property value to a monitored database instance.

property_id = 1
property_value_id = 1
database_id = 1
assign_property_value_url = f"{base_url}databases/" + str(database_id) + "/properties/" + str(property_id) + "/values/" + str(property_value_id)

try:
  print("\n*** Assigning custom property value ***")
  response = requests.post(assign_property_value_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  if response.status_code == 200:
  print(f"Custom property value assigned to the DB with ID: {database_id}")
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

#This will print out data like this:
#Custom property value assigned to the DB with ID: 1

Get all information about properties

This script returns information about all custom properties and their values.

get_properties_url = f"{base_url}databases/properties?require=assignment"

try:
  print("\n*** Getting custom property information ***")
  response = requests.get(get_properties_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

#This will print out data like this:
[
  {
    "id": 1,
    "name": "Location",
    "description": "Location of the database server",
    "values": [
      {
        "id": 1,
        "value": "New York",
        "assignment": [
          1
        ]
      }
    ],
    "unassigned": [
      2
    ]
  }
]

Delete a custom property

This script deletes a custom property.

property_id = 1
delete_property_url = f"{base_url}databases/properties/" + str(property_id)

try:
  print("\n*** Deleting custom property ***")
  response = requests.delete(delete_property_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  if response.status_code == 204:
  print(f"Custom property with ID of {property_id} deleted")
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

#This will print out data like this:
#Custom property with ID of 1 deleted

Full working script

The following script combines all of the examples shown above into a script that can be run.

import json
import sys
import time
import datetime
import requests

# ==========================================================
# Configure the variables below for the DPA Host
# ==========================================================
base_url = "http://localhost:8124/iwc/api/"
refresh_token = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJNeVRva2VuIiwiaXN..."
verify_cert = True
# ==========================================================

# ==========================================================
# Get Access Token
# ==========================================================
def get_access_header(prefix_url, rfrsh_token):
  """
  Given a base url and a refresh token retrieve the access token
  and return a header object with it.
  :param prefix_url: the base url
  :param rfrsh_token: refresh token used to get access token
  :return: the request header that contains the access token
  :rtype: dict
  """

  auth_token_url = prefix_url + "security/oauth/token"
  grant_type = "refresh_token"

  payload = {"grant_type": grant_type, "refresh_token": rfrsh_token}
  try:
    # get an access token
    resp = requests.post(auth_token_url, data=payload, verify=verify_cert)
    resp.raise_for_status()
    resp_json = resp.json()

    token_type = resp_json["token_type"]
    access_code = resp_json["access_token"]

    headers = {"authorization": f"{token_type} {access_code}",
		"content-type": "application/json;charset=UTF-8",
               "accept": "application/json"
    }

    return headers

  except requests.exceptions.HTTPError as ex:
    print(ex)
    print(ex.response.text)
    # print(json.dumps(json.loads(ex.response.text), indent=2))
    return None  # requests is bad return None, can't get access_code

# get the header that contains access token for authentication
header = get_access_header(base_url, refresh_token)
if header is None:
  sys.exit(0)

# ==========================================================
# Database Monitor Examples
# ==========================================================

# Calls for individual monitors...

# Get Monitor Information for a single database instance
database_id = 1
monitor_url = f"{base_url}databases/{database_id}/monitor-information"
single_monitor = None
try:
  print(f"\n*** Get Monitor Information for database with id of {database_id} ***")
  response = requests.get(monitor_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  single_monitor = response_json["data"]
  print(json.dumps(single_monitor, indent=2))

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Start or Stop monitoring a database instance given its database ID.
# If it is already running stop it and then restart it
# If it is not running start it and then stop it
if single_monitor is not None:
  monitor_url = f"{base_url}databases/{database_id}/monitor-status"
  if single_monitor["monitorState"] == "Monitor Running":
    change_command = "STOP"
    revert_command = "START"
  elif single_monitor["monitorState"] == "Monitor Stopped":
    change_command = "START"
    revert_command = "STOP"
  else:
    change_command = None
    revert_command = None

  if change_command is not None:
    try:
      print(f"\n*** {change_command} Monitor for database {database_id} ***")
      body = {"command": change_command}
      response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
      response.raise_for_status()
      response_json = response.json()
      print(json.dumps(response_json["data"], indent=2))

      print("Waiting 15 seconds...")
      time.sleep(15)

      print(f"\n*** {revert_command} Monitor for database {database_id} ***")
      body = {"command": revert_command}
      response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
      response.raise_for_status()
      response_json = response.json()
      print(json.dumps(response_json["data"], indent=2))

      print("Waiting 15 seconds...")
      time.sleep(15)

    except requests.exceptions.HTTPError as e:
      print(e)
      print(e.response.text)

# Calls for all monitors...

# Get Monitor Information for all database instances
database_id = 1
running_ids = []
monitor_url = f"{base_url}databases/monitor-information"
try:
  print("\n*** Get Information for a all database instances ***")
  response = requests.get(monitor_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  data = response_json["data"]
  print(json.dumps(data, indent=2))

  # Keep a list of running or started monitors to be used later
  for monitor in data:
    state = monitor["monitorState"]
    if state == "Monitor Running" or state == "Monitor Start No License" or 'Start' in state:
      running_ids.append(monitor["dbId"])

  print(f"Running Monitors: {running_ids}")

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Start monitoring all database instances.
monitor_url = f"{base_url}databases/monitor-status"
try:
  print("\n*** Starting all Monitors ***")
  body = {"command": "START"}
  response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
  print("Waiting 30 seconds...")
  time.sleep(30)

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Stop monitoring all database instances.
try:
  print("\n*** Stopping all Monitors ***")
  body = {"command": "STOP"}
  response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
  print("Waiting 30 seconds...")
  time.sleep(30)

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Try to put it back the way we found it by restarting the ones that were running
for db_id in running_ids:
  try:
    print(f"\n*** Starting Monitor for database {db_id} ***")
    monitor_url = f"{base_url}databases/{db_id}/monitor-status"
    body = {"command": "START"}
    response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
    response.raise_for_status()
    response_json = response.json()
    print(json.dumps(response_json["data"], indent=2))
  except requests.exceptions.HTTPError as e:
    print(e)
    print(e.response.text)

# Update the monitor database user password (Un-comment to use)
#monitor_url = f"{base_url}databases/{database_id}/update-password"
#try:
#  print(f"*** Update the Monitor password for database {database_id} ***")
#  body = {"password": "Password"}
#  response = requests.put(monitor_url, json=body, headers=header, verify=verify_cert)
#  response.raise_for_status()
#  response_json = response.json()
#  print(json.dumps(response_json["data"], indent=2))

#except requests.exceptions.HTTPError as e:
#  print(e)
#  print(e.response.text)

# ==========================================================
# Licensing Examples
# ==========================================================

# Get the currently installed license information
license_url = f"{base_url}databases/licenses/installed"
try:
  print("\n*** Getting Installed license information with total amounts available for use and total amounts used ***")
  response = requests.get(license_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  data = response_json["data"]
  print("licenseProduct licenseCategory licensesAvailable licensesConsumed")
  print("-------------- --------------- ----------------- ----------------")
  for i in range(len(data)):
    print('{:<15s}{:<16s}{:>17d}{:>17d}'.format(data[i]["licenseProduct"], data[i]["licenseCategory"],
    data[i]["licensesAvailable"], data[i]["licensesConsumed"]))

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Get License Information for a single database
license_url = f"{base_url}databases/{database_id}/licenses"
license_info = None
try:
  print(f"\n*** Getting current license information for the database instance with database ID of {database_id} ***")
  response = requests.get(license_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  license_info = response_json["data"]
  print(json.dumps(license_info, indent=2))

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# This will Update License Information for a single database setting the
# Performance License and the VM License to what it currently is.
# It should succeed but it should make no changes.
if license_info is not None:
  database_id = 1
  license_url = f"{base_url}databases/{database_id}/licenses"
  db_product = license_info["performanceLicenseProduct"]
  vm_product = license_info["vmLicenseProduct"]
  body = {"performanceLicenseProduct": db_product,
    "vmLicenseProduct": vm_product}
try:
  print(f"\n*** Updating license for database id {database_id} ***")
  response = requests.put(license_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))

except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# ==========================================================
# Annotation Examples
# ==========================================================

# Gets a List of annotations for the last 30 days
annotation_url = f"{base_url}databases/{database_id}/annotations"

# Dates are in ISO 8601 format ( 2018-12-31T12:00:00.000-07:00 )
end_time = datetime.datetime.now()
start_time = end_time + datetime.timedelta(days=-30)
args = {"startTime": start_time.astimezone().isoformat(),
        "endTime": end_time.astimezone().isoformat()}

try:
  print("\n*** Getting Annotations for the last 30 days ***")
  response = requests.get(annotation_url, params=args, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  print(json.dumps(response_json["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Create a new annotation
annotation_url = f"{base_url}databases/{database_id}/annotations"
annotation_id = None

#Dates are in ISO 8601 format, no millis ( 2018-12-31T12:00:00-07:00 )
create_time = datetime.datetime.now().replace(microsecond=0)
body = {"title": "API Test Title",
        "description": "API Test Description",
        "createdBy": "Test API User",
        "time": create_time.astimezone().isoformat()}
try:
  print("\n*** Creating Annotation ***")
  response = requests.post(annotation_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  data = response_json["data"]
  annotation_id = data["id"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Delete an annotation
if annotation_id is not None:
  annotation_url = f"{base_url}databases/{database_id}/annotations/{annotation_id}"
  try:
    print(f"\n*** Deleting Annotation with id of {annotation_id} ***")
    response = requests.delete(annotation_url, headers=header, verify=verify_cert)
    response.raise_for_status()
    if response.status_code == 204:
      print(f"Annotation with id of {annotation_id} deleted")
  except requests.exceptions.HTTPError as e:
    print(e)
    print(e.response.text)

# ==========================================================
# Registration Examples
# ==========================================================

# ----------------------------------------------------------
# Register a SQL Server database instance for monitoring.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/register-monitor"
body = {"databaseType": "SQLSERVER",
        "serverName": "127.0.0.1",
        "port": "1433",
        "sysAdminUser": "User",
        "sysAdminPassword": "Password",
        "monitoringUser": "dpa_test_m",
        "monitoringUserPassword": "Password",
        "monitoringUserIsNew": True,
        "displayName": "DPA_SQL2K12"}

new_db_id = None
try:
  print("\n*** Register SQL Server database ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  data = responseJson["data"]
  new_db_id = data["databaseId"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

print("Waiting 60 seconds...")
time.sleep(60)

# ----------------------------------------------------------
# Un-register the SQL Server database instance.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/unregister-monitor"
body = {"databaseId": new_db_id,
        "removeMonitoringUser": True,
        "removeDatabaseObjects": True,
        "sysAdminUser": "User",
        "sysAdminPassword": "Password"}
try:
  print(f"\n*** Unregister SQL Server database [{new_db_id}] ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  print(json.dumps(responseJson["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# ----------------------------------------------------------
# Register an Oracle database instance for monitoring.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/register-monitor"
body = {"databaseType": "ORACLE",
        "serverName": "127.0.0.1",
        "serviceNameOrSID": "DPA_ORA11R1",
        "port": "1521",
        "sysAdminUser": "User",
        "sysAdminPassword": "Password",
        "sysPassword": "Password",
        "monitoringUser": "dpa_test_m",
        "monitoringUserPassword": "Password",
        "monitoringUserIsNew": True,
        "monitoringUserTableSpace": "USERS",
        "monitoringUserTempTableSpace": "TEMP",
        "oracleEBusinessEnabled": False,
        "displayName": "DPA_ORA11R1"}

new_db_id = None
try:
  print("\n*** Register Oracle database ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  data = responseJson["data"]
  new_db_id = data["databaseId"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

print("Waiting 60 seconds...")
time.sleep(60)

# ----------------------------------------------------------
# Un-register the Oracle database instance.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/unregister-monitor"
body = {"databaseId": new_db_id,
        "removeMonitoringUser": True,
        "removeDatabaseObjects": True,
        "sysAdminUser": "User",
        "sysAdminPassword": "Password"}
try:
  print(f"\n*** Unregister Oracle database [{new_db_id}] ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  print(json.dumps(responseJson["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# ----------------------------------------------------------
# Register a MySQL database instance for monitoring.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/register-monitor"
body = {"databaseType": "MYSQL",
        "serverName": "127.0.0.1",
        "port": "3306",
        "sysAdminUser": "User",
        "sysAdminPassword": "Password",
        "monitoringUser": "dpa_test_m",
        "monitoringUserPassword": "Password",
        "monitoringUserIsNew": True,
        "displayName": "DPA_MYSQL56"}

new_db_id = None
try:
  print("\n*** Register MySQL database ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  data = responseJson["data"]
  new_db_id = data["databaseId"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

print("Waiting 60 seconds...")
time.sleep(60)
# ----------------------------------------------------------
# Un-register the MySQL database instance.
# ----------------------------------------------------------
registration_url = f"{base_url}databases/unregister-monitor"
body = {"databaseId": new_db_id,
        "removeMonitoringUser": True,
        "removeDatabaseObjects": True,
        "sysAdminUser": "User",
        "sysAdminPassword": "Password"}
try:
  print(f"\n*** Unregister MySQL database [{new_db_id}] ***")
  response = requests.post(registration_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  responseJson = response.json()
  print(json.dumps(responseJson["data"], indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)
# ==========================================================
# Custom Property Examples
# ==========================================================

# Create custom property
property_name = "Location"
property_description = "Location of the database server"
create_property_url = f"{base_url}databases/properties"
body = {
  "name": property_name,
  "description": property_description
}

property_id = None

try:
  print("\n*** Creating custom property ***")
  response = requests.post(create_property_url, json=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  data = response_json["data"]
  property_id = data["id"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Create value of the custom property
property_value = "New York"
create_value_url = f"{base_url}databases/properties/" + str(property_id) + "/values"
body = property_value

property_value_id = None

try:
  print("\n*** Creating custom property value ***")
  response = requests.post(create_value_url, data=body, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  data = response_json["data"]
  property_value_id = data["id"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Assign property value to DB
assign_property_value_url = f"{base_url}databases/" + str(database_id) + "/properties/" + str(property_id) + "/values/" + str(property_value_id)

try:
  print("\n*** Assigning custom property value ***")
  response = requests.post(assign_property_value_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  if response.status_code == 200:
  print(f"Custom property value assigned to the DB with ID: {database_id}")
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Get all information about properties (including DB assignment)
get_properties_url = f"{base_url}databases/properties?require=assignment"

try:
  print("\n*** Getting custom property information ***")
  response = requests.get(get_properties_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  response_json = response.json()
  data = response_json["data"]
  print(json.dumps(data, indent=2))
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

# Delete custom property
delete_property_url = f"{base_url}databases/properties/" + str(property_id)

try:
  print("\n*** Deleting custom property ***")
  response = requests.delete(delete_property_url, headers=header, verify=verify_cert)
  response.raise_for_status()
  if response.status_code == 204:
  print(f"Custom property with id of {property_id} deleted")
except requests.exceptions.HTTPError as e:
  print(e)
  print(e.response.text)

The scripts are not supported under any SolarWinds support program or service. The scripts are provided AS IS without warranty of any kind. SolarWinds further disclaims all warranties including, without limitation, any implied warranties of merchantability or of fitness for a particular purpose. The risk arising out of the use or performance of the scripts and documentation stays with you. In no event shall SolarWinds or anyone else involved in the creation, production, or delivery of the scripts be liable for any damages whatsoever (including, without limitation, damages for loss of business profits, business interruption, loss of business information, or other pecuniary loss) arising out of the use of or inability to use the scripts or documentation.