[Montellese]

I wrote the following python code yesterday to be able to make json rpc calls over HTTP or TCP to run some test commands:

Code:

#!/usr/bin/python

import socket
import select
import json

class JsonRpcResponseError(Exception):

    def __init__(self, errorCode, message = ""):
        self.ErrorCode = errorCode
        self.Message = message

    def __str__(self):
        return "Error %d in Json RPC response: %s" % (self.ErrorCode, self.Message)

class JsonRpcProtocol(object):
    HTTP = 0
    TCP = 1

class JsonRpcConnection(object):

    def __init__(self, protocol, address, port, username = "", password = ""):
        if protocol != JsonRpcProtocol.HTTP and protocol != JsonRpcProtocol.TCP:
            raise ValueError("Invalid protocol (HTTP or TCP)")
        if len(address) <= 0:
            raise ValueError("Invalid address")
        if port < 0:
            raise ValueError("Invalid port")
            
        self.__protocol = protocol
        self.__address = address
        self.__port = port
        self.__username = username
        self.__password = password

        self.__con = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.__connected = False
        self.__id = 1

    def __del__(self):
        if self.__connected:
            self.__con.shutdown(socket.SHUT_RDWR)
            self.__con.close()

    def call(self, method, params = None):
        return self.__send(method, params)

    def __connect(self):
        if not self.__connected:
            socket.setdefaulttimeout(5.0)
            addr = self.__address
            if len(self.__username) > 0:
                addr = "@%s" % addr
                if len(self.__password) > 0:
                    addr = ":%s%s" % (self.__password, addr)
                addr = "%s%s" % (self.__username, addr)
            
            self.__con.connect((addr, self.__port))
            self.__connected = True

    def __send(self, method, params):
        self.__connect()
        request = "{\"jsonrpc\": \"2.0\", \"method\": \"%s\", " % method
        if not (params is None):
            request += "\"params\": %s, " % json.dumps(params)
            pass

        request += "\"id\": %d}" % self.__id
        
        self.__id += 1

        if self.__protocol == JsonRpcProtocol.HTTP:
            self.__con.send("POST /jsonrpc HTTP/1.1\x0D\x0A".encode("utf-8"))
            self.__con.send("\x0D\x0A".encode("utf-8"))

        self.__con.send(bytearray(request, 'utf-8'))
        
        response = ""
        while True:
            response += self.__con.recv(4096).decode("utf-8")

            if len(select.select([self.__con], [], [], 0)[0]) == 0:
                break

        return self.__parseResponse(response)

    def __parseResponse(self, response):
        if response.find("HTTP") == 0:
            response = response[response.find("\x0D\x0A\x0D\x0A") + 4:len(response)]
        
        jsonObj = json.loads(response)
        
        if "error" in jsonObj:
            raise JsonRpcResponseError(jsonObj["error"]["code"],
                                       jsonObj["error"]["message"])
        if "result" in jsonObj:
            return jsonObj["result"]
        
        raise JsonRpcResponseError("Invalid response")

I'm sure there are better ways to do this as this is basically my first python code I have ever written but maybe it will get you started.