GoldenGate REST API basics with Python

Oracle GoldenGate REST API has been around for quite some time now, but I’ve yet to see it used in practice at customers. Every time I tried to introduce it, it was some sort of novelty. Mainly because DBAs tend to dislike automation, but also for technical reasons. Even though the REST API was introduced in GoldenGate 12c with the Microservices Architecture, some customers are still stuck with the Classic Architecture. As a reminder, the classic architecture is now completely absent from the 23ai version (plan your migration now !).

That being said, where to start when using the GoldenGate REST API? Oracle has some basic documentation using curl, but I want to take things a bit further by leveraging the power of Python, starting with basic requests.

Make your first request to the GoldenGate REST API

If you really have no idea what a REST API is, there are tons of excellent articles online for you to get into it. Getting back to the basics, in Python, the requests module will handle the API calls for us.

The most basic REST API call would look like what I show below. Adapt the credentials, and the service manager host and port.

import requests

url = "http://vmogg:7809/services"
auth = ("ogg_user", "ogg_password")
result = requests.get(url, auth=auth)

Until now, nothing fascinating, but with the 200 return code below, we know that the call succeeded. The ok flag gives us the status of the call:

>>> result
<Response [200]>
>>> result.ok
True

And to get the real data returned by the API, use the json method.

>>> result.json()
{'$schema': 'api:versions', 'links': [{'rel': 'current', 'href': 'http://vmogg:7809/services/v2', 'mediaType': 'application/json'}, {'rel': 'canonical', 'href': 'http://vmogg:7809/services', 'mediaType': 'application/json'}, {'rel': 'self', 'href': 'http://vmogg:7809/services', 'mediaType': 'application/json'}], 'items': [{'$schema': 'api:version', 'version': 'v2', 'isLatest': True, 'lifecycle': 'active', 'catalog': {'links': [{'rel': 'canonical', 'href': 'http://vmogg:7809/services/v2/metadata-catalog', 'mediaType': 'application/json'}]}}]}

Up until that point, you are successfully making API connections to your GoldenGate service manager, but nothing more. What you need to change is the URL, and more specifically the endpoint.

/services is called an endpoint, and the full list of endpoints can be found in the GoldenGate documentation. Not all of them are useful, but when looking for a specific GoldenGate action, this endpoint library is a good starting point.

For instance, to get the list of all the deployments associated with your service manager, use the /services/v2/deployments endpoint. If you get an OGG-12064 error, it means that the credentials are not correct (they were technically not needed for the first example).

>>> url = "http://vmogg:7809/services/v2/deployments"
>>> requests.get(url, auth=auth).json()
{
    '$schema': 'api:standardResponse',
    'links': [
        {'rel': 'canonical', 'href': 'http://vmogg:7809/services/v2/deployments', 'mediaType': 'application/json'},
        {'rel': 'self', 'href': 'http://vmogg:7809/services/v2/deployments', 'mediaType': 'application/json'},
        {'rel': 'describedby', 'href': 'http://vmogg:7809/services/v2/metadata-catalog/versionDeployments', 'mediaType': 'application/schema+json'}
    ],
    'messages': [],
    'response': {
        '$schema': 'ogg:collection',
        'items': [
            {'links': [{'rel': 'parent', 'href': 'http://vmogg:7809/services/v2/deployments', 'mediaType': 'application/json'}, {'rel': 'canonical', 'href': 'http://vmogg:7809/services/v2/deployments/ServiceManager', 'mediaType': 'application/json'}], '$schema': 'ogg:collectionItem', 'name': 'ServiceManager', 'status': 'running'},
            {'links': [{'rel': 'parent', 'href': 'http://vmogg:7809/services/v2/deployments', 'mediaType': 'application/json'}, {'rel': 'canonical', 'href': 'http://vmogg:7809/services/v2/deployments/ogg_test_01', 'mediaType': 'application/json'}], '$schema': 'ogg:collectionItem', 'name': 'ogg_test_01', 'status': 'running'},
            {'links': [{'rel': 'parent', 'href': 'http://vmogg:7809/services/v2/deployments', 'mediaType': 'application/json'}, {'rel': 'canonical', 'href': 'http://vmogg:7809/services/v2/deployments/ogg_test_02', 'mediaType': 'application/json'}], '$schema': 'ogg:collectionItem', 'name': 'ogg_test_02', 'status': 'running'}
        ]
    }
}

Already, we’re starting to get a bit lost in the output (even though I cleaned it for you). Without going too much into the details, when the call succeeds, we are interested in the response.items object, discarding $schema and links objects. When the call fails, let’s just display the output for now.

def parse(response):
    try:
        return response.json()
    except ValueError:
        return response.text

def extract_main(result):
    if not isinstance(result, dict):
        return result
    resp = result.get("response", result)
    if "items" not in resp:
        return resp
    exclude = {"links", "$schema"}
    return [{k: v for k, v in i.items() if k not in exclude} for i in resp["items"]]

result = requests.get(url, auth=auth)
if result.ok:
    response = parse(result)
    main_response = extract_main(response)

We now have a more human-friendly output for our API calls ! For this specific example, we only retrieve the deployment names and their status.

>>> main_response
[{'name': 'ServiceManager', 'status': 'running'}, {'name': 'ogg_test_01', 'status': 'running'}, {'name': 'ogg_test_02', 'status': 'running'}]

GoldenGate POST API calls

Some API calls require you to send data instead of just receiving it. A common example is the creation of a GoldenGate user. Both the role and the username are part of the endpoint. Using the post method instead of get, we will give the user settings (the password, essentially) in the params argument:

import requests

role = "User"
user = "ogg_username"
url = f"http://vmogg:7809/services/v2/authorizations/{role}/{user}"
auth = ("ogg_user", "ogg_password")
data = {
    "credential": "your_password"
}

result = requests.post(url, auth=auth, json=data)

To check if the user was created, you can go to the Web UI or check the ok flag again.

>>> result.ok
True

Here, the API doesn’t provide us with much information when the call succeeds:

>>> result.text
'{"$schema":"api:standardResponse","links":[{"rel":"canonical","href":"http://vmogg:7809/services/v2/authorizations/User/ogg_username","mediaType":"application/json"},{"rel":"self","href":"http://vmogg:7809/services/v2/authorizations/User/ogg_username","mediaType":"application/json"}],"messages":[]}'

A fully working OGGRestAPI Python class

When dealing with the REST API, you will quickly feel the need for a standard client object that will handle everything for you. A very basic ogg_rest_api.py script class will look like this:

import requests
import urllib3


class OGGRestAPI:
    def __init__(self, host, port=7809, username=None, password=None,
                 protocol='http', ca_cert=None, verify_ssl=True):
        """
        Initialize Oracle GoldenGate REST API client

        :param host: hostname or IP
        :param port: port number
        :param username: service username
        :param password: service password
        :param protocol: 'http' or 'https'
        :param ca_cert: path to a trusted CA cert (for self-signed certs)
        :param verify_ssl: bool, whether to verify SSL certs
        """
        self.base_url = f'{protocol}://{host}:{port}'
        self.auth = (username, password)
        self.headers = {'Accept': 'application/json', 'Content-Type': 'application/json'}
        self.verify_ssl = ca_cert if ca_cert else verify_ssl

        if not verify_ssl and protocol == 'https':
            # Disable InsecureRequestWarning if verification is off
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

    def _request(self, method, path, *, params=None, data=None, extract=True):
        url = f'{self.base_url}{path}'
        response = requests.request(
            method,
            url,
            auth=self.auth,
            headers=self.headers,
            params=params,
            json=data,
            verify=self.verify_ssl  # <-- use CA cert or bool
        )
        self._check_response(response)
        result = self._parse(response)
        return self._extract_main(result) if extract else result

    def _get(self, path, params=None, extract=True):
        return self._request('GET', path, params=params, extract=extract)

    def _post(self, path, data=None, extract=True):
        return self._request('POST', path, data=data, extract=extract)

    def _put(self, path, data=None, extract=True):
        return self._request('PUT', path, data=data, extract=extract)

    def _patch(self, path, data=None, extract=True):
        return self._request('PATCH', path, data=data, extract=extract)

    def _delete(self, path, extract=True):
        return self._request('DELETE', path, extract=extract)

    def _check_response(self, response):
        if not response.ok:
            print(f'HTTP {response.status_code}: {response.text}')
            response.raise_for_status()

    def _parse(self, response):
        try:
            return response.json()
        except ValueError:
            return response.text

    def _extract_main(self, result):
        if not isinstance(result, dict):
            return result

        resp = result.get('response', result)
        if 'items' not in resp:
            return resp

        exclude = {'links', '$schema'}
        return [{k: v for k, v in i.items() if k not in exclude} for i in resp['items']]

With this, we can connect to the API and generate the same GET query as before to retrieve all deployments. This time, we only provide the endpoint, and not the whole URL.

>>> from ogg_rest_api import OGGRestAPI
>>> client_blog = OGGRestAPI(host="vmogg", port=7809, protocol="http", username="ogg", password="ogg")
>>> client_blog._get("/services/v2/deployments")
[{'name': 'ServiceManager', 'status': 'running'}, {'name': 'ogg_test_01', 'status': 'running'}, {'name': 'ogg_test_02', 'status': 'running'}]

As you can imagine, all GoldenGate API functionalities can be integrated in this class, enhancing GoldenGate management and monitoring. Next time you want to automate your GoldenGate processes, please consider using this REST API !

REST API calls to a secured GoldenGate deployment

If your GoldenGate deployment is secured, you can still use this Python class. The requests module will handle it for you. I give two examples below for a secured deployment using a self-signed certificate:

# Checking CA automatically (Trusted CA)
>>> client_blog = OGGRestAPI(host="vmogg", port=7809, protocol="https", username="ogg", password="ogg")

# Providing RootCA (self-signed certificate)
>>> client_blog = OGGRestAPI(host="vmogg", port=7809, protocol="https", username="ogg", password="ogg", ca_cert="/u01/app/oracle/ogg_certs/152-67-87-131/RootCA_cert.pem", verify_ssl=True)

# Disabling verification
>>> client_blog = OGGRestAPI(host="vmogg", port=7809, protocol="https", username="ogg", password="ogg", verify_ssl=False)