1. Why I Built This

Every team that integrates with Netcool/OMNIbus ObjectServer eventually writes the same thing: a chunk of Python that wrangles JayDeBeApi, figures out the right Sybase driver class name, constructs JDBC URLs, handles primary/backup failover, and translates raw cursor rows into usable dictionaries. I needed all of that for the Netcool Connector test automation framework, so I built it once, properly, as a reusable module.

If your team tests any connector that reads from or writes to an ObjectServer — or if you need to script alert management tasks — this client saves you that work entirely.

2. How It Works

The client wraps JDBC connectivity using JayDeBeApi, a Python library that bridges Python code to Java JDBC drivers via JPype. Under the hood, it speaks the Sybase TDS protocol to ObjectServer on port 4100 (the default ObjectServer AGG_P port). You give it a config dict and paths to the two required JARs; it handles the rest.

The class is one file, zero project-specific dependencies. It only imports from the Python standard library plus jaydebeapi. You can copy it directly into any project.

3. Prerequisites

java -version          # verify
brew install openjdk   # macOS
sudo yum install java-11-openjdk  # RHEL/CentOS

cp /opt/IBM/tivoli/netcool/omnibus/java/jars/nco_g_java.jar  python/jars/
cp /opt/IBM/tivoli/netcool/omnibus/java/jars/jconn3.jar      python/jars/

git clone git@github.ibm.com:katamari/test-automation/test-automation-netcool-connector.git
ls test-automation-netcool-connector/python/jars/
# nco_g_java.jar  jconn3.jar  ...

pip install jaydebeapi JPype1
# or: pip install -r python/requirements.txt

nc -zv <objectserver-host> 4100

4. Installation

Clone the repo — the client and both JARs are already in place:

git clone git@github.ibm.com:katamari/test-automation/test-automation-netcool-connector.git
cd test-automation-netcool-connector
pip install -r python/requirements.txt

Or drop just the client into an existing project:

cp python/netcool_objectserver_client.py  /your/project/
cp python/jars/nco_g_java.jar             /your/project/jars/
cp python/jars/jconn3.jar                 /your/project/jars/

5. Configuration

config = {
    "username": "root",
    "password": "netcool_password",
    "primary_objectserver": {
        "url": "objectserver.example.com",
        "api_port": 4100
    },
    # Optional — client retries here if primary fails
    "backup_objectserver": {
        "url": "objectserver-backup.example.com",
        "api_port": 4100
    },
    "jdbc_driver_path": [
        "python/jars/nco_g_java.jar",
        "python/jars/jconn3.jar"
    ],
    "tls": False
}

import os
config = {
    "username":  os.environ["NETCOOL_USER"],
    "password":  os.environ["NETCOOL_PASSWORD"],
    "primary_objectserver": {
        "url":      os.environ["NETCOOL_HOST"],
        "api_port": int(os.environ.get("NETCOOL_PORT", "4100"))
    },
    "jdbc_driver_path": ["python/jars/nco_g_java.jar", "python/jars/jconn3.jar"]
}

6. Connecting and Disconnecting

from netcool_objectserver_client import NetcoolObjectServerClient

client = NetcoolObjectServerClient(config)
try:
    client.connect()
    # ... work ...
finally:
    client.disconnect()   # always runs, even on exception

connect() tries primary first, then backup if configured. It probes three Sybase driver class variants (jConnect 2.x, 3.x, 4.x) automatically, so it works regardless of which version is in your JAR.

7. The 11 Built-In Examples

The main() function contains 11 commented-out examples covering every common operation. To run one, uncomment the relevant block, update the config with your ObjectServer details, then:

python python/netcool_objectserver_client.py

8. Alert Operations — Full API Reference

8.1 Query alerts with a filter

alerts = client.get_alerts("Severity = 5 AND Acknowledged = 0", limit=50)
for alert in alerts:
    print(f"Serial={alert['Serial']}  Node={alert['Node']}  Summary={alert['Summary']}")

8.2 Look up by Serial

alert = client.get_alert_by_serial(12345)   # returns Dict or None

8.3 Get all alerts for a node

node_alerts = client.get_alerts_by_node("prod-router-01.example.com")

8.4 Alert counts by severity

counts = client.get_alert_count_by_severity()   # {5: 12, 4: 7, 3: 21, ...}

8.5 Insert a test alert

inserted = client.insert_test_alert(
    node="test-node-01", summary="CPU high", severity=4, agent="PythonClient"
)
print(f"Serial={inserted['Serial']}")   # auto-generated, returned immediately

8.6 – 8.9 Modify / remove

client.update_alert_severity(serial=12345, new_severity=5)
client.acknowledge_alert(serial=12345)    # Acknowledged=1, StateChange updated
client.clear_alert(serial=12345)          # Severity=0
client.delete_alert(serial=12345)         # hard DELETE

8.10 Arbitrary SQL

# SELECT
rows = client.execute_query("""
    SELECT TOP 10 Serial, Node, Summary, Severity
    FROM alerts.status WHERE Manager = 'ProbeWatch'
    ORDER BY LastOccurrence DESC
""")

# INSERT / UPDATE / DELETE
client.execute_update(
    "UPDATE alerts.status SET Acknowledged = 1 WHERE AlertGroup = 'TestGroup'"
)

9. How the Client Powers the BDD Scenario: New Alert Inserted in ObjectServer

This is the first — and currently most complete — scenario in outbound_flow.feature. It validates the full outbound path from ObjectServer all the way to Kafka:

The Gherkin scenario reads:

Scenario: New Alert Inserted in ObjectServer
  When 4 new alerts are inserted in ObjectServer with:
    | Field    | Value                  |
    | Node     | TestNode               |
    | Summary  | Test alert from Python |
    | Severity | 5                      |
    | Agent    | test                   |
  Then all 4 alerts should be successfully inserted in ObjectServer
  And each inserted alert should have a Serial number
  And the gateway should detect the INSERT operations
  And the connector should receive "insertAlert" events
  And at least 1 ChangeRequest should be created with type "create"
  And the ChangeRequest should contain:
    | Field                | Value                  |
    | entityType           | alert                  |
    | entity.summary       | Test alert from Python |
    | entity.severity      | 6                      |
    | entity.state         | open                   |
    | entity.resource.name | TestNode               |
  And at least 1 ChangeRequest should be published to Kafka as a CloudEvent

The step definition for "4 new alerts are inserted" calls insert_test_alert() four times, collecting each returned Serial into context.current_alerts. The Kafka verification step then builds a set of those Serial integers and filters the full topic history — up to 100,000 messages — looking for a ChangeRequest whose entity.sender.serverSerial is in that set:

# Built from the Serial numbers returned by insert_test_alert()
inserted_serials = {alert.get("Serial") for alert in context.current_alerts if alert.get("Serial")}

# O(1) per-message membership test — efficient even at 100k messages
matching_requests = [
    msg for msg in messages
    if (msg.get("type") == "create"
        and msg.get("entityType") == "alert"
        and msg.get("entity", {}).get("sender", {}).get("serverSerial") in inserted_serials)
]

10. API Quick Reference

11. Troubleshooting

12. How to Get It

git clone git@github.ibm.com:katamari/test-automation/test-automation-netcool-connector.git

python/
├── netcool_objectserver_client.py   ← the client (with 11 examples in main())
├── jars/
│   ├── nco_g_java.jar               ← Netcool gateway library
│   └── jconn3.jar                   ← Sybase JDBC driver
├── SOLUTION.md                      ← connectivity options incl. REST API alternative
├── PYTHON_CLIENT_README.md          ← full usage reference
└── TROUBLESHOOTING.md               ← detailed error diagnosis

If you run into something the troubleshooting table does not cover, or if you extend the client with new methods your team needs, open a PR or ping me on Slack.