Testing the MQ REST API

  • Post category:MQProgramming
  • Reading time:3 mins read

I have been playing around with the MQ REST API. It works very well. Also certificate-based authentication work out of the box.

Of course, you are doing something that MQ-fanatics might find horrific: reliable messaging over an unreliable protocol. They are somewhat right. By no means can MQ provide assured message delivery over an unreliable HTTP protocol. When using this in application, make sure you handle all error situations. For example, when you do not get an http response, you don’t know whether the message was successfully delivered or not. You application has to cater for such situations. Some call this idempotence.

Here is my small Python program that illustrates how you can use the MQ REST API.

import requests
import json
import sys

class MQWebManager:
    baseapiurl = "/ibmmq/rest/v1/messaging"
    
    def __init__(self, ep, ak, cert_file_path, key_file_path):
        self.endpoint = ep
        self.apikey = ak
        self.cert = (cert_file_path, key_file_path) 

    def apideleterequest(self, qmgr, queue, msgid):
        # operation = POST or DELETE
        resourceurl = self.endpoint + "/ibmmq/rest/v1/messaging/qmgr/" + qmgr + "/queue/" + queue + "/message"
        request_headers = {
            'messageId': "'" + msgid + "'",
            'Content-Type' : 'text/plain;charset=utf-8' ,
            'ibm-mq-rest-csrf-token' : 'somevalue',
            'correlationId' : ''
            }
        data = {}
        response = requests.delete(resourceurl, data=data, cert=self.cert, verify=False, headers=request_headers)
        return response

    def apipostrequest(self, qmgr, queue):
        # operation = POST or DELETE
        resourceurl = self.endpoint + "/ibmmq/rest/v1/messaging/qmgr/" + qmgr + "/queue/" + queue + "/message"
        request_headers = {
            'Content-Type' : 'text/plain;charset=utf-8' ,
            'ibm-mq-rest-csrf-token' : 'somevalue'
            }
        data = 'hello from apipostrequest'
        print('resource url: ', resourceurl)
        response = requests.post(resourceurl, data=data, cert=self.cert, verify=False, headers=request_headers)
        return response



print('---------')

#cert_file_path = "/yourpath/yourcert.crt"   
#key_file_path = "/yourpath/yourcert.privkey"

cert_file_path = sys.argv[1]
key_file_path = sys.argv[2]

m1 = MQWebManager("https://mqweb.yourzos.com:12345","", cert_file_path, key_file_path)

#put a message on the queue
response = m1.apipostrequest("QMGR","YOUR.Q.NAME") 
print(">>>", response.status_code, response.json)

print(response.headers) 

print(response)
#retrieve msgid from the message we just put there
msgid = response.headers['ibm-mq-md-messageId']
print(response.headers['ibm-mq-md-messageId'])

#delete that message we just put there
response = m1.apideleterequest("QMGR","YOUR.Q.NAME", msgid) 
print(">>>", response.status_code, response.json)



print('---------')

Try out you IBM Z Open Automation Utilities

  • Post category:Programming
  • Reading time:1 mins read

Please find a mini Python program to check out your IBM Z Open Automation Utilities installation and explore its possibilities.

Find more information in the IBM documentation.

Be aware of the particular working of passing parameters to a Python program through the command line.

#import logging_config
import sys
from zoautil_py import datasets 


# total arguments
n = len(sys.argv)
print("Total arguments passed: ", n)

def printname(dsn):
    print("Testing python and IZOAU")
    print("Dataset " + dsn + " exists: " + str(datasets.exists(dsn)))

if (n > 2):
    print("Pass 1 parameter (dataset name) or less (use default dataset name)") 
elif (n==1):
    printname("MYDATA.DATASET")
else:
    printname(sys.argv[2])

Invoke with:

python testibmzoau.py DATASET.NAME