Production-ready GoldenGate REST client in Python

In this blog, I provide GoldenGate administrators with a fully working, production-ready Python client to efficiently handle GoldenGate REST API calls. From extract creation to user management, it has never been easier to manage and monitor your GoldenGate deployments !

Basic OGGRestAPI Python class

In a previous blog post, I presented a basic Python script for making calls to the GoldenGate REST API. I also provided an OGGRestAPI Python class to handle the connections and API responses. This class serves as the base structure of the client.

import requests
import urllib3


class OGGRestAPI:
    def __init__(self, url, username=None, password=None, ca_cert=None, verify_ssl=True,
                 test_connection=False, timeout=None):
        """
        Initialize Oracle GoldenGate REST API client.

        :param url: Base URL of the OGG REST API. It can be:
                    'http(s)://hostname:port' without NGINX reverse proxy,
                    'https://nginx_host:nginx_port' with NGINX reverse proxy.
        :param username: service username
        :param password: service password
        :param ca_cert: path to a trusted CA cert (for self-signed certs)
        :param verify_ssl: bool, whether to verify SSL certs
        :param test_connection: if True, will attempt to retrieve API versions on init
        :param timeout: request timeout in seconds
        """
        self.base_url = url
        self.username = username
        self.swagger_version = '2026.01.27'
        self.auth = (self.username, password)
        self.headers = {'Accept': 'application/json', 'Content-Type': 'application/json'}
        self.verify_ssl = ca_cert if ca_cert else verify_ssl
        self.timeout = timeout
        self.session = requests.Session()
        self.session.auth = self.auth
        self.session.headers.update(self.headers)

        if not verify_ssl and self.base_url.startswith('https://'):
            # Disable InsecureRequestWarning if verification is off
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

        # Optional connection check
        if test_connection:
            try:
                self._request('GET', '/services', extract=False)
                print(f'Connected to OGG REST API at {self.base_url}')
            except Exception as e:
                print(f'Error connecting to OGG REST API: {e}')
                raise

    def _request(self, method, path, *, params=None, data=None, extract=True):
        url = f'{self.base_url}{path}'
        response = self.session.request(
            method,
            url,
            auth=self.auth,
            headers=self.headers,
            params=params,
            json=data,
            verify=self.verify_ssl,
            timeout=self.timeout
        )
        self._check_response(response)
        result = self._parse(response)
        return self._extract_main(result) if extract else result

    def _build_path(self, template, path_params=None):
        path_params = path_params or {}
        return template.format(**path_params)

    def _call(self, method, template, *, path_params=None, params=None, data=None, extract=True):
        path = self._build_path(template, path_params=path_params)
        result = self._request(method, path, params=params, data=data, extract=False)
        if extract:
            return self._extract_main(result)
        return 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:
            if 'messages' in response.json():
                messages = response.json().get('messages', [])
                raise Exception(
                    ' ; '.join([f"{message['severity']}: {message['title']}" for message in messages])
                )
            else:
                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 close(self):
        self.session.close()

    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']]

While this is already useful when starting with GoldenGate automation, searching for the correct endpoints and methods or providing the correct parameters can quickly become daunting. That is why I decided to build and release an easy-to-use Python client for the GoldenGate REST API.

To make things easier for developers, REST APIs often come with some sort of standard template file that will guide the user on how to query the API and interpret the results. Oracle provides a swagger.json file, which you can get for GoldenGate 26ai and 19c.

There are multiple ways online to transform these Swagger files into working Python clients, but they are very often not tailored to the specific needs of the API, and unusable in practice. Hence my proposal with this blog post.

Example of a REST API method in the client

There are around 300 methods included in the client that I provide. Here is an example with the create_alias method.

# Endpoint: /services/{version}/authorizations/{role}/{user}
def create_user(self, user, role, data=None, version='v2'):
    """
    POST /services/{version}/authorizations/{role}/{user}
    Required Role: Security
    Create a new Authorization User Resource.

    Parameters:
        user (string): User Resource Name Example: user_example
        role (string): Authorization Role Resource Name Example: User
        version (string): Oracle GoldenGate Service API version. Example: v2
        body (object):  Example: body_example

    Example:
        client.create_user(
            user='user_example',
            role='User',
            data={
                "credential": "password-A1",
                "info": "Credential Information"
            })
    """
    return self._call(
        "POST",
        "/services/{version}/authorizations/{role}/{user}",
        path_params={"user": user, "role": role, "version": version},
        data=data,
    )

With this method, you can easily create an alias in GoldenGate, with the following code. But more on that later.

Where to find the code ?

The full code is available in this GitHub repository. Feel free to use it in your deployments. I provided one Python class for GoldenGate 26ai, one for 23ai (if you are still using it) and another one for GoldenGate 19c. For the most basic functions, there is not much difference between the clients, but for more advanced usage, I would suggest using the code dedicated to the proper version of GoldenGate.

How to use the client ?

Administering your GoldenGate deployments with this client is straightforward. Just start by importing the class and creating a client instance.

from oggrestapi import OGGRestAPI

# Connect to http://vmogg:7809 with user 'ogg'
ogg_client = OGGRestAPI(host="vmogg", port=7809, username="ogg", password="ogg", protocol="http")

The connection status is displayed when the client is initialized:

>>> from oggrestapi import OGGRestAPI
>>> ogg_client = OGGRestAPI(url="http://vmogg:7809", username="ogg", password="ogg")
Connected to OGG REST API at http://vmogg:7809

To illustrate the benefits of using this client, here is what a standard API call to create a GoldenGate user would look like with Python.

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)

The same operation using the Python client becomes:

ogg_client.create_user(
    user="ogg_username",
    role="User",
    data={"credential": "your_password"}
)

All endpoint parameters (such as user and role, in the example above) are now method arguments (create_user, in this case). To keep things simple and not over-engineer, the data payload of the API is kept. But depending on the endpoint, you could easily add wrapper functions to further simplify usage.

Call examples

I give below a few other basic calls that you can use. You should pay attention to the port you are using when making calls to the API. Some endpoints are only accessible on specific services. For instance, list_extracts will only work on the administration service of a deployment (default port 7810), while list_deployments only works on the service manager (default port 7809).

• List deployments associated with a service manager.

>>> ogg_client.list_deployments()
[{'name': 'ServiceManager', 'status': 'running'}, {'name': 'ogg_test_01', 'status': 'running'}]

• List extracts in a deployment (for this one, ogg_client has to be created on the administration service of the deployment if you are not using an NGINX proxy).

>>> ogg_client.list_extracts()
[{'name': 'EXT01', 'status': 'stopped'}, {'name': 'EXT02', 'status': 'running'}, {'name': 'EXT03', 'status': 'running'}]

• List and retrieve tasks

>>> ogg_client.list_tasks()
[{'name': 'purge_aa'}]

>>> ogg_client.retrieve_task('purge_aa')
{'enabled': True, 'critical': False, 'status': 'stopped', 'command': {'name': 'purge', 'purgeType': 'trails', 'useCheckpoints': True, 'trails': [{'name': 'aa', 'path': 'PDB1'}], 'keep': [{'type': 'min', 'value': 48, 'units': 'hours'}]}, 'schedule': {'every': {'units': 'days', 'value': 1}}, 'restart': {'enabled': False}, '$schema': 'ogg:task'}

And of course, you can create and manage extracts, replicats, etc. through this client. All tasks available through the API are replicated in the client.

Method naming explanation

I only wrote this section of the blog for those who would like to dig a bit deeper. I generated the method names using three different layers:

• The basic method name is the summary given by Oracle. For instance, the endpoint given below is associated with the create_alias method. This way, the vast majority of methods written in the Python script have a very simple and canonical name that is easy to remember or to retrieve from the documentation. Words like a, an or of are removed from the name.

• Unfortunately, some methods still have the same summary. I give an example below with the “Retrieve Status” summary, which is available for both extracts and replicats. To avoid having two methods named retrieve_status, I resolve those duplicate methods and add the parameters that are unique to the methods. In this specific case, the retrieve_status method is not created, but there is one retrieve_status_extract and one retrieve_status_replicat method.

• More than 99% of methods are covered by the first two cases, allowing for a very efficient and clear naming. When it’s not the case, like the “Get a list of distribution paths” summary below, I search for the difference in REST API endpoint path and add the difference. Here, we get two methods named get_list_distribution_paths_targets and get_list_distribution_paths_sources.

In upcoming blog posts, I will provide concrete examples on how to manage your GoldenGate deployments from A to Z with this client. In the meantime, do not hesitate if you have any feedback on this little project !