aiothornode

aiothornode

This is a simple Python library to access THORChain nodes. It is asynchronous and uses aiohttp.

Important

v.0.1.0 breaking changes:

• NamedTuples instead of DataClasses for Thor entities
• Environments were renamed
• No more consensus features

v.0.0.21 is hotfix. Port 1317 was disabled, so it is an emergency upgrade to save this lib. More news will be later...

Features:

• Now it is just a convenient wrapper for THORNode API

Supported endpoints:

• Constants
• Mimir
• Nodes (node accounts)
• Current TX queue length
• Pools (current and at arbitrary height)
• Tendermint block at height
• Inbound addresses and other chain info
• Asgard & Yggdrasil vaults (new!)
• Balance of THOR account

Installation

python -m pip install git+https://github.com/tirinox/aiothornode

Quick start

The following code is quite self-documenting:

import random

from aiothornode.connector import *
from aiothornode.env import *
import asyncio
import aiohttp


def delim():
    print('-' * 100)


async def main():
    env = MAINNET.copy()
    # env = ThorEnvironment(...)  # custom

    async with aiohttp.ClientSession() as session:
        connector = ThorConnector(env, session)

        genesis = await connector.query_genesis()
        print(f'Chain ID = {genesis["chain_id"]}')
        delim()

        chains = await connector.query_chain_info()
        chains = list(chains.values())
        print('Chain info:', chains)
        delim()

        print('Tendermint Block:')
        tender_block = await connector.query_tendermint_block_raw(100001)
        block_header = tender_block['result']['block']['header']
        print(f'{block_header["height"] = } and {block_header["time"] = }')
        delim()

        constants = await connector.query_constants()
        print(f'Constants: {constants}')
        delim()

        mimir = await connector.query_mimir()
        mimir_1 = mimir.get('MINIMUMBONDINRUNE')
        print(f'Mimir: {mimir}, MINIMUMBONDINRUNE = {mimir_1}')
        delim()

        queue = await connector.query_queue()
        print(f'Queue: {queue}')
        delim()

        node_accounts = await connector.query_node_accounts(consensus=False)
        print(f'Example node account: {random.sample(node_accounts, 1)[0]}')
        delim()

        pool = await connector.query_pool('BNB.BUSD-BD1', height=8218339)
        print(pool)
        delim()

        pools = await connector.query_pools()
        print(pools[0])
        print(f'Total {len(pools)} pools')
        delim()

        bank = await connector.query_balance('thor1dheycdevq39qlkxs2a6wuuzyn4aqxhve4qxtxt')
        print(f'Balance of {bank.address} is {bank.runes_float} Rune')
        delim()


if __name__ == '__main__':
    asyncio.run(main())

Testing

Install PyTest and an async plugin for it:

pip install pytest
pip install pytest-asyncio

Then run

pytest test