Cannot save message information to my database

I am currently using Elektron feed to get real-time information about prices for different RIC codes. The sample code I got from this link https://github.com/Refinitiv/websocket-api/tree/master/Applications/Examples/python

I added extra modules and modified slightly a code:

_send_market_price_request(self, ric_name) - Inserted RIC codes manually

_on_message(self, message) - I added part which should save the message information to my database

When I do not add part which saves information to my db, I can see JSON messages in the output console from this line

print(message_json)

However, if I modify it as:

print(message_json)
cur.execute("""INSERT INTO ENK_OP.ENKadhoc (DataDateTime,Name,MyValueOne) 
VALUES (cast(TIMESTAMP 'NOW' as date), '{0}', '{1}')"""
    .format(
    message_json[0]['Key'].get('Name', 'Unknown'),
    1000

))

cur.execute("commit;")

I cannot see any output from print(message_json) and the code starts relogin/get new token every 10 sec.

Could you help to identify why this problem arises, as I believe the save to my db code part should work and does not affect print() function.

My python code is available here: ELEKTRON.txt

Find more posts tagged with

python

elektron

websockets

rrto

refinitiv-realtime-optimised

Accepted answers

Jirapongse

@g.suhharukov

I have run the code. The first problem is "ric" is not defined when calling the self._send_market_price_request method.

Therefore, I change it to the below.

    def _send_market_price_request(self):
        """ Create and send simple Market Price request """
        mp_req_json = {
            'ID': 2,
            'Key': {
                'Name': [
                    'JPY=', 'THB=', 'EUR='

                ],
            },
...
    def _process_login_response(self, message_json):
        """ Send item request """
        if message_json['State']['Stream'] != "Open" or message_json['State']['Data'] != "Ok":
            print("Login failed.")
            sys.exit(1)

        self.logged_in = True
        self._send_market_price_request()

Next, some messages (such as status messages) may not have the 'Key' property. Therefore, it is better to verify it before accessing its value.

    def _on_message(self, message):
        """ Called when message received, parse message into JSON for processing """
        print("RECEIVED on " + self.session_name + ":")
        #print(message)
        message_json = json.loads(message)
        print(message_json)
        print("Message Key")
        if('Key' in message_json[0]):
            print(message_json[0]['Key'].get('Name', 'Unknown'))
        print("Message Key End")

All comments

Jirapongse

@g.suhharukov

It may relate to the capital X letter in the code.

    for x in range(len(message_json)):
            cur.execute("""INSERT INTO MYDB (DATE,Name,MyValue) 
            VALUES (cast(TIMESTAMP 'NOW' as date), '{0}', '{1}')"""
                .format(
                message_json[X]['Key'].get('Name', 'Unknown'),
                1000

            ))

The for loop uses the lower case (x) but in the code block it uses the capital one (X).

g.suhharukov

Yes, indeed, there is a typo, should be lowercase 'x'. Additionally, if I try to write a message_json[0] (without loop) in order to take the first part of the message the same problem arises.

Jirapongse

@g.suhharukov

I have run the code. The first problem is "ric" is not defined when calling the self._send_market_price_request method.

Therefore, I change it to the below.

    def _send_market_price_request(self):
        """ Create and send simple Market Price request """
        mp_req_json = {
            'ID': 2,
            'Key': {
                'Name': [
                    'JPY=', 'THB=', 'EUR='

                ],
            },
...
    def _process_login_response(self, message_json):
        """ Send item request """
        if message_json['State']['Stream'] != "Open" or message_json['State']['Data'] != "Ok":
            print("Login failed.")
            sys.exit(1)

        self.logged_in = True
        self._send_market_price_request()

Next, some messages (such as status messages) may not have the 'Key' property. Therefore, it is better to verify it before accessing its value.

    def _on_message(self, message):
        """ Called when message received, parse message into JSON for processing """
        print("RECEIVED on " + self.session_name + ":")
        #print(message)
        message_json = json.loads(message)
        print(message_json)
        print("Message Key")
        if('Key' in message_json[0]):
            print(message_json[0]['Key'].get('Name', 'Unknown'))
        print("Message Key End")

g.suhharukov

@jirapongse.phuriphanvichai

Yes, I wrote that in my code I used several RIC codes, so the RIC codes you provided also suit as an example. Yes, your modified print works well, and to save the data I also tried to slightly modify the code as

if ('Key' in message_json[0]):

   cur.execute("INSERT INTO ENK_OP.ENKtemptrades (DATE,Name,MyValue)VALUES (cast(TIMESTAMP 'NOW' as date), '{0}', '{1}')".format(message_json[0]['Key'].get('Name', 'Unknown'),1000))

print(message_json)

So it should save 'Name' from Key, however, the same problem again (no information is saved and no messages received). Usually, under 'RECEIVED on session1' I have JSON messages (so many times this 'RECEIVED on' repeats, but in this case only once).

RECEIVED on session1: RECEIVED on session1: [{'Type': 'Ping'}] SENT on session1: {
  "Type":"Pong" }
RECEIVED on session1: [{'Type': 'Ping'}] SENT on session1: {
  "Type":"Pong" }

Continue in the next comment....

g.suhharukov

From the code above I identified that these Ping-Pong messages come from

def _process_message(self, message_json):   message_type = message_json['Type']

I tried to simply use print(message_json['Type']), but see the same behavior that it does not print anything from messages at all

def _on_message(self, message):
    print("RECEIVED on " + self.session_name + ":")
    message_json = json.loads(message)

    print(message_json['Type'])

EXPLORE OUR SITES