from setuptools import setup, find_packages

with open("README.md", "r") as f:
    page_description = f.read()

with open("requirements.txt") as f:
    requirements = f.read().splitlines()


setup(
    name="tcabci_read_client",
    version="v0.1.5",
    author="transferchain",
    description="transferchain abci read client",
    long_description=page_description,
    long_description_content_type="text/markdown",
    url="https://github.com/transferchain/tcabci-read-python-client",
    packages=find_packages(),
    install_requires=requirements,
    python_requires='>=3.8',
)

setup.py
TransferChain/tcabci-read-python-client
from .client import HttpClient, WsClient, MessageType


__all__ = [
    HttpClient,
    WsClient,
    MessageType
]

tcabci_read_client/__init__.py
TransferChain/tcabci-read-python-client
import json
import logging
import threading
from collections import namedtuple
from urllib.parse import urljoin
from enum import Enum
import requests
import websocket


VERSION = 'v0.1.5'

logger = logging.getLogger('tcabci-read-client')
logger.setLevel(logging.DEBUG)

ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)

formatter = logging.Formatter(
    '%(asctime)s - '
    '%(name)s - %(levelname)s - %(message)s')

ch.setFormatter(formatter)
logger.addHandler(ch)


class MessageType(Enum):
    SUBSCRIBE = 'subscribe'
    UNSUBSCRIBE = 'unsubscribe'


HttpResultTuple = namedtuple(
    'HttpResultTuple', 'success error_message result')


class HttpClient:

    def __init__(self, http_url, http_headers=None):
        self.http_url = http_url
        if http_headers is None:
            http_headers = {
                'Client': f'tcabaci-read-python-client-{VERSION}'
            }
        self.http_headers = http_headers

    def get_last_block(self):
        url = urljoin(self.http_url, 'v1/blocks?limit=1&offset=0')
        req = requests.get(url, headers=self.http_headers)
        if not req.ok:
            return HttpResultTuple(
                success=False,
                error_message='invalid status code', result=req.content)
        try:
            response = req.json()
        except Exception as e:
            logger.error(e)
            return HttpResultTuple(
                success=False, error_message=str(e), result=req.content)

        result = {"blocks": response["data"],
                  "total_count": response["total_count"]}
        return HttpResultTuple(
            success=True, error_message=None, result=result)

    def broadcast(self, tx_id, version, fee, data, sign, tx_type,
                  sender_address, recipient_address):
        transaction = {
            "fee": fee,
            "id": tx_id,
            "version": version,
            "data": data,
            "sign": sign,
            "type": tx_type,
            "sender_addr": sender_address,
            "recipient_addr": recipient_address
        }
        payload = json.dumps(transaction)
        url = urljoin(self.http_url, 'v1/broadcast')
        req = requests.post(url, data=payload, headers=self.http_headers)
        if not req.ok:
            logger.error("invalid status code")
            return HttpResultTuple(
                success=False, error_message='invalid status code',
                result=req.content)
        try:
            response = req.json()
        except Exception as e:
            logger.error(e)
            return HttpResultTuple(
                success=False, error_message=str(e), result=req.content)
        return HttpResultTuple(
            success=True, error_message=None, result=response)

    def tx_search(self, *args, **kwargs):
        pre_payload = {
            "recipient_addrs": kwargs.get('recipientAddrs'),
            "sender_addrs": kwargs.get('senderAddrs'),
            "hashes": kwargs.get('hashes'),
            "typ": kwargs.get('typ'),
            "limit": kwargs.get('limit'),
            "offset": kwargs.get('offset'),
            "order_field": kwargs.get('orderField'),
            "order_by": kwargs.get('orderBy')
        }
        height_operator = kwargs.get('heightOperator')
        height = kwargs.get('height')
        if height_operator and height:
            pre_payload['height'] = f"{height_operator} {height}"

        payload = {k: v for k, v in pre_payload.items() if v is not None}
        url = urljoin(self.http_url, 'v1/tx_search/p')
        req = requests.post(url, json=payload, headers=self.http_headers)
        if req.status_code == 400:
            logger.error('invalid arguments')
            return HttpResultTuple(
                success=False, error_message='invalid arguments',
                result=req.content)

        if not req.ok:
            logger.error("invalid status code")
            return HttpResultTuple(
                success=False, error_message='invalid status code',
                result=req.content)

        try:
            response = req.json()
        except Exception as e:
            logger.error(e)
            return HttpResultTuple(
                success=False, error_message=str(e), result=req.content)

        result = {"txs": response["data"],
                  "total_count": len(response["data"])}
        return HttpResultTuple(
            success=True, error_message=None, result=result)


class WsClient(object):

    def __init__(self, ws_url, message_callback, error_callback):
        self.ws_url = ws_url
        self.message_callback = message_callback
        self.error_callback = error_callback

        self.__listen = None
        self.websocket = None
        self.subscribed_addresses = set()

    def listener(self):
        logger.debug('Websocket listener is starting')
        while self.__listen is None:
            try:
                message = self.websocket.recv()
                self.message_callback(message)
            except Exception as e:
                self.error_callback(e)

    def ws(self):
        self.websocket = websocket.create_connection(self.ws_url)
        logger.debug("WebSocket connection opened")

    def start(self):
        logger.debug("Websocket client starting")
        self.ws()
        thd = threading.Thread(target=self.listener, daemon=True)
        thd.start()

    def stop(self):
        self.websocket.close()
        self.websocket = None
        self.__listen = False
        self.subscribed_addresses = set()
        logger.debug("WebSocket connection closed")

    def subscribe(self, address):
        address = set(address) - self.subscribed_addresses
        payload = json.dumps({
            "IsWeb": True,
            "Type": MessageType.SUBSCRIBE.value,
            "Addrs": list(address)
        })

        self.websocket.send(payload)
        self.subscribed_addresses.update(address)

    def unsubscribe(self, address=None):
        if not address:
            address = self.subscribed_addresses
        address = set(address)
        self.subscribed_addresses = self.subscribed_addresses - address
        payload = json.dumps({
            "IsWeb": True,
            "Type": MessageType.UNSUBSCRIBE.value,
            "Addrs": list(address)
        })
        self.websocket.send(payload)

    def get_subscribe_addresses(self):
        return self.subscribed_addresses

tcabci_read_client/client.py
TransferChain/tcabci-read-python-client
import unittest
import json
import time
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
from tcabci_read_client import HttpClient


class DummyHandler(BaseHTTPRequestHandler):

    def do_POST(self):
        print(self.path)
        if self.path == '/v1/tx_search/p':
            pass
        elif self.path == '/v1/broadcast':
            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            response = {"data":
                        {
                            "hash": "E08F7AD",
                            "code": 0,
                            "data": "",
                            "log": "",
                            "codespace": ""},
                        "total_count": 0,
                        "error": True,
                        "errors": None,
                        "detail": "Created"}
            self.wfile.write(json.dumps(response).encode('utf-8'))
        else:
            self.send_response(404)
            self.send_header('Content-type', 'text/html')
            self.end_headers()
            self.wfile.write(b'Not found!')

    def do_GET(self):
        if self.path == '/v1/blocks?limit=1&offset=0':
            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            response = {
                "data": [
                    {
                        "id": 159724,
                        "inserted_at": "2023-09-29T14:56:34.592463Z",
                        "height": 151533,
                        "txs": 1,
                        "hash": "36EE7A43201D0B20A60A730A352B55CD6BC73B815872F20DC8D881497439D9B5",  # noqa
                        "transactions": None
                    }
                ],
                "total_count": 151514,
                "error": False,
                "errors": None,
                "detail": "OK"
            }
            self.wfile.write(json.dumps(response).encode('utf-8'))
        elif self.path == 'v1/tx_search/p':
            pass
        elif self.path == 'v1/broadcast':
            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            response = {"data":
                        {
                            "hash": "E08F7AD",
                            "code": 0,
                            "data": "",
                            "log": "",
                            "codespace": ""},
                        "total_count": 0,
                        "error": True,
                        "errors": None,
                        "detail": "Created"}
            self.wfile.write(json.dumps(response).encode('utf-8'))
        else:
            self.send_response(404)
            self.send_header('Content-type', 'text/html')
            self.end_headers()
            self.wfile.write(b'Not found!')


class TestWebSocketClient(unittest.TestCase):
    HOST = 'localhost'
    PORT = 9000

    def start_dummy_http_server(self):
        server = HTTPServer((self.HOST, self.PORT), DummyHandler)
        t = threading.Thread(target=server.serve_forever, daemon=True)
        t.start()
        print('dummy http server is running.')

    def test_client(self):
        self.start_dummy_http_server()
        time.sleep(1)
        client = HttpClient(f"http://{self.HOST}:{self.PORT}")
        rsp = client.get_last_block()
        self.assertEqual(True, rsp.success)

        rsp = client.broadcast(
            tx_id=1, version=1, fee=0, data="", sign="", tx_type="",
            sender_address="from", recipient_address="to")
        self.assertEqual(True, rsp.success)


if __name__ == '__main__':
    unittest.main()

tests/test_httpclient.py
TransferChain/tcabci-read-python-client
import unittest
import json
import time
import asyncio
import threading
import websockets
from tcabci_read_client import WsClient


def simple_ws_server(host, port, loop):
    asyncio.set_event_loop(loop)

    async def echo(websocket, path):
        async for message in websocket:
            await websocket.send(message)

    start_server = websockets.serve(echo, host, port)
    try:
        loop.run_until_complete(start_server)
        asyncio.get_event_loop().run_forever()
    except OSError:
        pass


class TestWebSocketClient(unittest.TestCase):

    def test_client(self):
        host = 'localhost'
        port = 8000
        loop = asyncio.new_event_loop()
        t = threading.Thread(target=simple_ws_server,
                             args=(host, port, loop),
                             daemon=True)
        t.start()
        time.sleep(1)

        def message_callback(message):
            if not message:
                return
            message_json = json.loads(message)
            self.assertEqual(message_json['Addrs'][0], 'test')

        def error_callback(error):
            pass

        ws_client = WsClient(
            f"ws://{host}:{port}/ws",
            message_callback=message_callback,
            error_callback=error_callback)
        ws_client.start()
        ws_client.subscribe(['test'])
        ws_client.unsubscribe(['test'])
        time.sleep(1)
        ws_client.stop()


if __name__ == '__main__':
    unittest.main()

tests/test_websocket.py
TransferChain/tcabci-read-python-client