TonX 0.1.5

A user-friendly Python library for interacting with TON (The Open Network), offering convenient payment handling and much more.

TonX

TonX is an asynchronous Tonlib wrapper for TON community written in Python.

Features

• Easy, Fast and Powerful
• Fully asynchronous
• Supports Payments, Wallet Tracker, Tonlib types/functions and much more.

Requirements

NOTE: TonX will automatically attempt to download the correct pre-built binary of tonlibjson from Tonlib Releases

• Python 3.9+

Installation

You can install TonX using pip:

pip install tonx

To install with payments support, use:

pip install tonx[payments]

To install the development version from Github, use the following command:

pip install git+https://github.com/AYMENJD/tonx.git

Examples

Basic usage

Getting wallet balance of Fragment:

from tonx import Client, types, utils
from urllib.request import urlopen
import json, asyncio


async def main():
    ton_config = json.loads(
        urlopen("https://ton-blockchain.github.io/global.config.json").read()
    )  # Main net

    async with Client(
        config=ton_config,
        keystore=types.KeyStoreTypeDirectory("tonlib-db/"),
    ) as client:
        account_state = await client.getAccountState(
            types.AccountAddress("EQBAjaOyi2wGWlk-EDkSabqqnF-MrrwMadnwqrurKpkla9nE")
        )

        if account_state.getType() != "error":
            balance_in_toncoin = utils.from_nanograms(account_state.balance)

            print(
                f"Fragment.com wallet has {utils.truncate_zeros(balance_in_toncoin)} TON"
            )
        else:
            print(
                f"Something went wrong: {account_state.code} - {account_state.message}"
            )


asyncio.run(main())

Payments

Accepting and handling payments in TON:

from tonx import Client, types, utils
from tonx.payments import Payments, Invoice
from urllib.request import urlopen
import json, asyncio, logging

logging.basicConfig(
    level=logging.INFO,
    format="[%(levelname)s][p %(process)d %(threadName)s][%(created)f][%(filename)s:%(lineno)d][%(funcName)s]  %(message)s",
)


async def main():
    # ton_config = json.loads(
    #     urlopen("https://ton-blockchain.github.io/global.config.json").read()
    # ) # Main net
    ton_config = json.loads(
        urlopen("https://ton-blockchain.github.io/testnet-global.config.json").read()
    )  #  Test net

    client = Client(
        config=ton_config,
        keystore=types.KeyStoreTypeDirectory("tonlib-db/"),
    )
    payments = Payments(
        client=client,
        account_address=types.AccountAddress(
            "WALLET_ADDRESS"  # Wallet address that will receive the payments
        ),
    )

    @payments.on_invoicePayment()
    async def invoice_payment_handler(client, invoice: Invoice):
        if invoice.is_completed:
            payment_info = (
                f"Payment for Invoice #{invoice.id}\n"
                f"Amount: {utils.truncate_zeros(invoice.paid_amount)} TON\n"
                f"From: {utils.Address.normalize(invoice.paid_by_address)}\n"
                f"Is completed: {invoice.is_completed}\n"
                f"Extra: {invoice.extra}\n\n"
            )
        elif invoice.is_expired:
            payment_info = (
                f"Payment for expired Invoice #{invoice.id}\n"
                f"Amount: {utils.truncate_zeros(invoice.paid_amount)} TON\n"
                f"From: {utils.Address.normalize(invoice.paid_by_address)}\n"
                f"Is completed: {invoice.is_completed}\n"
                f"Extra: {invoice.extra}\n\n"
            )
        print(payment_info)

    await client.start(payments)

    # Create an example invoice
    invoice_1 = await payments.createInvoice(
        amount=0.05,
        comment="This is a test payment for TonX\n\n",
        extra={
            "name": "AYMEN",
            "user_id": 1088394097,
        },  # Extra data that will be stored with the invoice
        ttl=None,  # Invoice is valid for every with no expire time (NOTE: the default value for ttl is 3600 seconds a.k.a 1 hour)
    )

    invoice_2 = await payments.createInvoice(
        0.05,
        "This is a test payment for TonX\n\n",
        extra={"name": "AWM", "user_id": 39809485},
        ttl=300,  #  Invoice is valid for 5 minutes (the minimum value)
    )

    print(f"Payment link for invoice 1 ({invoice_1.id}): {invoice_1.payment_link}")
    print(f"Payment link for invoice 2 ({invoice_2.id}): {invoice_2.payment_link}")
    print()

    await client.idle(register_signal_handlers=True)


asyncio.run(main())

Tracking

An example of Tracking a wallet transactions, for example @Wallet Bot

from tonx import Client, types, utils
from tonx.payments import Payments
from urllib.request import urlopen
import json, logging

logging.basicConfig(
    level=logging.INFO,
    format="[%(levelname)s][p %(process)d %(threadName)s][%(created)f][%(filename)s:%(lineno)d][%(funcName)s]  %(message)s",
)

ton_config = json.loads(
    urlopen("https://ton-blockchain.github.io/global.config.json").read()
)  # Main net

client = Client(
    config=ton_config,
    keystore=types.KeyStoreTypeDirectory("tonlib-db/"),
)
payments = Payments(
    client=client,
    account_address=types.AccountAddress(
        "EQBDanbCeUqI4_v-xrnAN0_I2wRvEIaLg1Qg2ZN5c6Zl1KOh"  # Wallet to watch transactions, for example t.me/Wallet address
    ),
)


@payments.on_incomingTransaction(
    filter_fn=lambda _, transaction: utils.from_nanograms(transaction.in_msg.value)
    >= 1  # Handle only transaction that is 1+ TON
)
async def handle_incoming_transactions(_, transaction: types.RawTransaction):
    comment = None
    if transaction.in_msg.msg_data.getType() == "msg.dataText":
        comment = transaction.in_msg.msg_data.text.decode()

    transaction_info = (
        f"Received a transaction:\n"
        f"Amount: `{utils.truncate_zeros(utils.from_nanograms(transaction.in_msg.value))} TON`\n"
        f"From: {utils.Address.normalize(transaction.in_msg.source)}\n"
        f"Comment: {comment}\n"
    )
    print(transaction_info)


@payments.on_outgoingTransaction()
async def handle_outgoing_transactions(_, transaction: types.RawTransaction):
    comment = None
    if transaction.out_msgs[0].msg_data.getType() == "msg.dataText":
        comment = transaction.out_msgs[0].msg_data.text.decode()

    transaction_info = (
        f"Sending "
        f"`{utils.truncate_zeros(utils.from_nanograms(transaction.out_msgs[0].value))} TON` "
        f"to `{utils.Address.normalize(transaction.out_msgs[0].destination)}` "
        f"({comment})\n"
    )
    print(transaction_info)


client.run(payments)

License

MIT License