puffpastry-backend/app/contracts/contract.py

import asyncio

from stellar_sdk import Keypair, Network, SorobanServer, TransactionBuilder, xdr as stellar_xdr
from stellar_sdk.exceptions import PrepareTransactionException
from stellar_sdk.soroban_rpc import GetTransactionStatus, SendTransactionStatus
from stellar_sdk.xdr import SCVal, SCValType


class SingletonMeta(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            instance = super().__call__(*args, **kwargs)
            cls._instances[cls] = instance
        return cls._instances[cls]


class SmartContract(metaclass=SingletonMeta):
    contract_id: str
    user_key: str
    def __init__(self, contract_id: str, user_key: str):
        self.contract_id = contract_id
        self.user_key = user_key

    async def _execute_procedure(self, procedure_name: str, parameters=None):
        source_keypair = Keypair.from_secret(self.user_key)

        soroban_server = SorobanServer('https://soroban-testnet.stellar.org')

        source_account = soroban_server.load_account(source_keypair.public_key)

        built_transaction = (
            TransactionBuilder(
                source_account=source_account,
                base_fee=100,
                network_passphrase=Network.TESTNET_NETWORK_PASSPHRASE,
            )
            .append_invoke_contract_function_op(
                contract_id=self.contract_id,
                function_name=procedure_name,
                parameters=parameters,
            )
            .set_timeout(30)
            .build()
        )

        try:
            prepared_transaction = soroban_server.prepare_transaction(built_transaction)
        except PrepareTransactionException as e:
            print(f"Exception preparing transaction: {e}\n{e.simulate_transaction_response.error}")
            raise e

        prepared_transaction.sign(source_keypair)

        send_response = soroban_server.send_transaction(prepared_transaction)

        if send_response.status != SendTransactionStatus.PENDING:
            raise Exception("sending transaction failed")

        while True:
            get_response = soroban_server.get_transaction(send_response.hash)
            if get_response.status != GetTransactionStatus.NOT_FOUND:
                break
            await asyncio.sleep(2)

        print(f"get_transaction response: {get_response}")

        if get_response.status == GetTransactionStatus.SUCCESS:
            assert get_response.result_meta_xdr is not None

            transaction_meta = stellar_xdr.TransactionMeta.from_xdr(
                get_response.result_meta_xdr
            )
            return_value = transaction_meta.v3.soroban_meta.return_value
            output = translate_soroban_value(return_value)
            return output
        else:
            print(f"Transaction failed: {get_response.result_xdr}")


def translate_soroban_value(val: SCVal) -> int or str or bool or list[int or str or bool] or None:
    def sanitize(k: str) -> str:
        return k.lstrip('b\'').rstrip('\'')

    type_handlers = {
        SCValType.SCV_U32: lambda v: v.u32.uint32,
        SCValType.SCV_I32: lambda v: v.i32.int32,
        SCValType.SCV_U64: lambda v: v.u64.uint64,
        SCValType.SCV_I64: lambda v: v.i64.int64,
        SCValType.SCV_BOOL: lambda v: v.bool.boolean,
        SCValType.SCV_STRING: lambda v: sanitize(str(v.str.sc_string)),
        SCValType.SCV_SYMBOL: lambda v: sanitize(str(v.sym.sc_symbol)),
        SCValType.SCV_BYTES: lambda v: bytes(v.bytes.sc_bytes),
        SCValType.SCV_VEC: lambda v: [translate_soroban_value(item) for item in v.vec.sc_vec],
        SCValType.SCV_MAP: lambda v: {translate_soroban_value(item.key): translate_soroban_value(item.val) for item in v.map.sc_map},
    }

    if val.type == SCValType.SCV_VOID:
        return None

    handler = type_handlers.get(val.type)
    if handler:
        return handler(val)
    else:
        raise ValueError(f"Unsupported SCVal type: {val.type}")