diff --git a/ton-http-api/pyTON/api/api_v2/endpoints/common.py b/ton-http-api/pyTON/api/api_v2/endpoints/common.py index 6670db8..9cbf0e3 100644 --- a/ton-http-api/pyTON/api/api_v2/endpoints/common.py +++ b/ton-http-api/pyTON/api/api_v2/endpoints/common.py @@ -14,9 +14,34 @@ from fastapi.exceptions import HTTPException from pyTON.schemas import ( - TonResponse, + TonResponse, DeprecatedTonResponseJsonRPC, - TonRequestJsonRPC + get_get_address_information_error_responses, + GetAddressInformationResponse, + GetExtendedAddressInformationResponse, + GetWalletInformationResponse, + GetAddressBalanceResponse, + GetAddressStateResponse, + PackAddressResponse, + UnpackAddressResponse, + TonGetTokenDataResponse, + TonResponse200Generic, + GetConfigParamResponse, + MasterchainInfo, + MasterchainSignatures, + ShardBlockProof, + BlockId, + ConsensusBlock, + Shards, + BlockHeader, + RunGetMethodResponse, + AddressForms, + SendBocReturnHashResponse, + EstimateFeeResponse, + OkResponse, + TonRequestJsonRPC, + ShortTransactions, + TransactionId, ) from pyTON.core.tonlib.manager import TonlibManager from pyTON.api.deps.ton import tonlib_dep, settings_dep @@ -24,11 +49,15 @@ from tvm_valuetypes.cell import deserialize_cell_from_object -from pytonlib.utils.address import detect_address as __detect_address, prepare_address as _prepare_address +from pytonlib.utils.address import ( + detect_address as __detect_address, + prepare_address as _prepare_address, +) from pytonlib.utils.wallet import wallets as known_wallets, sha256 from loguru import logger +from pyTON.schemas.ton import RawTransaction router = APIRouter() settings = settings_dep() @@ -50,7 +79,10 @@ def prepare_address(address): def address_state(account_info): - if isinstance(account_info.get("code", ""), int) or len(account_info.get("code", "")) == 0: + if ( + isinstance(account_info.get("code", ""), int) + or len(account_info.get("code", "")) == 0 + ): if len(account_info.get("frozen_hash", "")) == 0: return "uninitialized" else: @@ -61,13 +93,17 @@ def address_state(account_info): def wrap_result(func): @wraps(func) async def wrapper(*args, **kwargs): - result = await asyncio.wait_for(func(*args, **kwargs), settings.tonlib.request_timeout) + result = await asyncio.wait_for( + func(*args, **kwargs), settings.tonlib.request_timeout + ) return TonResponse(ok=True, result=result) + return wrapper json_rpc_methods = {} + def json_rpc(method): def g(func): @wraps(func) @@ -77,7 +113,7 @@ def f(**kwargs): # Add function's default value parameters to kwargs. if k not in kwargs and v.default is not inspect._empty: default_val = v.default - + if isinstance(default_val, Param) or isinstance(default_val, Body): if default_val.default == ...: raise TypeError("Non-optional argument expected") @@ -87,7 +123,9 @@ def f(**kwargs): # Some values (e.g. lt, shard) don't fit in json int and can be sent as str. # Coerce such str to int. - if (v.annotation is int or v.annotation is Optional[int]) and type(kwargs[k]) is str: + if (v.annotation is int or v.annotation is Optional[int]) and type( + kwargs[k] + ) is str: try: kwargs[k] = int(kwargs[k]) except ValueError: @@ -97,16 +135,25 @@ def f(**kwargs): json_rpc_methods[method] = f return func + return g -@router.get('/getAddressInformation', response_model=TonResponse, response_model_exclude_none=True, tags=['accounts']) -@json_rpc('getAddressInformation') +@router.get( + "/getAddressInformation", + response_model=TonResponse200Generic[GetAddressInformationResponse], + responses=get_get_address_information_error_responses(), + response_model_exclude_none=True, + tags=["accounts"], +) +@json_rpc("getAddressInformation") @wrap_result async def get_address_information( - address: str = Query(..., description="Identifier of target TON account in any form."), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Query( + ..., description="Identifier of target TON account in any form." + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Get basic information about the address: balance, code, data, last_transaction_id. """ @@ -115,17 +162,29 @@ async def get_address_information( result["state"] = address_state(result) if "balance" in result and int(result["balance"]) < 0: result["balance"] = 0 - if result["sync_utime"] < 1803189600 and _detect_address(address)["raw_form"] in suspended_accounts: + if ( + result["sync_utime"] < 1803189600 + and _detect_address(address)["raw_form"] in suspended_accounts + ): result["suspended"] = True return result -@router.get('/getExtendedAddressInformation', response_model=TonResponse, response_model_exclude_none=True, tags=['accounts']) -@json_rpc('getExtendedAddressInformation') + +@router.get( + "/getExtendedAddressInformation", + response_model=TonResponse200Generic[GetExtendedAddressInformationResponse], + responses=get_get_address_information_error_responses(), + response_model_exclude_none=True, + tags=["accounts"], +) +@json_rpc("getExtendedAddressInformation") @wrap_result async def get_extended_address_information( - address: str = Query(..., description="Identifier of target TON account in any form."), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Query( + ..., description="Identifier of target TON account in any form." + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Similar to previous one but tries to parse additional information for known contract types. This method is based on tonlib's function *getAccountState*. For detecting wallets we recommend to use *getWalletInformation*. """ @@ -133,21 +192,38 @@ async def get_extended_address_information( result = await tonlib.generic_get_account_state(address) return result -@router.get('/getWalletInformation', response_model=TonResponse, response_model_exclude_none=True, tags=['accounts']) -@json_rpc('getWalletInformation') + +@router.get( + "/getWalletInformation", + response_model=TonResponse200Generic[GetWalletInformationResponse], + responses=get_get_address_information_error_responses(), + response_model_exclude_none=True, + tags=["accounts"], +) +@json_rpc("getWalletInformation") @wrap_result async def get_wallet_information( - address: str = Query(..., description="Identifier of target TON account in any form."), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Query( + ..., description="Identifier of target TON account in any form." + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Retrieve wallet information. This method parses contract state and currently supports more wallet types than getExtendedAddressInformation: simple wallet, standart wallet, v3 wallet, v4 wallet. """ address = prepare_address(address) result = await tonlib.raw_get_account_state(address) - res = {'wallet': False, 'balance': 0, 'account_state': None, 'wallet_type': None, 'seqno': None} + res = { + "wallet": False, + "balance": 0, + "account_state": None, + "wallet_type": None, + "seqno": None, + } res["account_state"] = address_state(result) - res["balance"] = result["balance"] if (result["balance"] and int(result["balance"]) > 0) else 0 + res["balance"] = ( + result["balance"] if (result["balance"] and int(result["balance"]) > 0) else 0 + ) if "last_transaction_id" in result: res["last_transaction_id"] = result["last_transaction_id"] ci = sha256(result["code"]) @@ -158,31 +234,72 @@ async def get_wallet_information( wallet_handler["data_extractor"](res, result) return res -@router.get('/getTransactions', response_model=TonResponse, response_model_exclude_none=True, tags=['accounts', 'transactions']) -@json_rpc('getTransactions') + +@router.get( + "/getTransactions", + response_model=TonResponse200Generic[List[RawTransaction[TransactionId]]], + response_model_exclude_none=True, + tags=["accounts", "transactions"], +) +@json_rpc("getTransactions") @wrap_result async def get_transactions( - address: str = Query(..., description="Identifier of target TON account in any form."), - limit: Optional[int] = Query(default=10, description="Maximum number of transactions in response.", gt=0, le=100), - lt: Optional[int] = Query(default=None, description="Logical time of transaction to start with, must be sent with *hash*."), - hash: Optional[str] = Query(default=None, description="Hash of transaction to start with, in *base64* or *hex* encoding , must be sent with *lt*."), - to_lt: Optional[int] = Query(default=0, description="Logical time of transaction to finish with (to get tx from *lt* to *to_lt*)."), - archival: bool = Query(default=False, description="By default getTransaction request is processed by any available liteserver. If *archival=true* only liteservers with full history are used."), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Query( + ..., description="Identifier of target TON account in any form." + ), + limit: Optional[int] = Query( + default=10, + description="Maximum number of transactions in response.", + gt=0, + le=100, + ), + lt: Optional[int] = Query( + default=None, + description="Logical time of transaction to start with, must be sent with *hash*.", + ), + hash: Optional[str] = Query( + default=None, + description="Hash of transaction to start with, in *base64* or *hex* encoding , must be sent with *lt*.", + ), + to_lt: Optional[int] = Query( + default=0, + description="Logical time of transaction to finish with (to get tx from *lt* to *to_lt*).", + ), + archival: bool = Query( + default=False, + description="By default getTransaction request is processed by any available liteserver. If *archival=true* only liteservers with full history are used.", + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Get transaction history of a given address. """ address = prepare_address(address) - return await tonlib.get_transactions(address, from_transaction_lt=lt, from_transaction_hash=hash, to_transaction_lt=to_lt, limit=limit, archival=archival) - -@router.get('/getAddressBalance', response_model=TonResponse, response_model_exclude_none=True, tags=['accounts']) -@json_rpc('getAddressBalance') + return await tonlib.get_transactions( + address, + from_transaction_lt=lt, + from_transaction_hash=hash, + to_transaction_lt=to_lt, + limit=limit, + archival=archival, + ) + + +@router.get( + "/getAddressBalance", + response_model=GetAddressBalanceResponse, + responses=get_get_address_information_error_responses(), + response_model_exclude_none=True, + tags=["accounts"], +) +@json_rpc("getAddressBalance") @wrap_result async def get_address_balance( - address: str = Query(..., description="Identifier of target TON account in any form."), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Query( + ..., description="Identifier of target TON account in any form." + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Get balance (in nanotons) of a given address. """ @@ -192,13 +309,22 @@ async def get_address_balance( result["balance"] = 0 return result["balance"] -@router.get('/getAddressState', response_model=TonResponse, response_model_exclude_none=True, tags=['accounts']) -@json_rpc('getAddressState') + +@router.get( + "/getAddressState", + response_model=GetAddressStateResponse, + responses=get_get_address_information_error_responses(), + response_model_exclude_none=True, + tags=["accounts"], +) +@json_rpc("getAddressState") @wrap_result async def get_address( - address: str = Query(..., description="Identifier of target TON account in any form."), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Query( + ..., description="Identifier of target TON account in any form." + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Get state of a given address. State can be either *unitialized*, *active* or *frozen*. """ @@ -206,212 +332,332 @@ async def get_address( result = await tonlib.raw_get_account_state(address) return address_state(result) -@router.get('/packAddress', response_model=TonResponse, response_model_exclude_none=True, tags=['accounts']) -@json_rpc('packAddress') + +@router.get( + "/packAddress", + response_model=PackAddressResponse, + responses=get_get_address_information_error_responses(), + response_model_exclude_none=True, + tags=["accounts"], +) +@json_rpc("packAddress") @wrap_result async def pack_address( - address: str = Query(..., description="Identifier of target TON account in raw form.", example="0:83DFD552E63729B472FCBCC8C45EBCC6691702558B68EC7527E1BA403A0F31A8"), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Query( + ..., + description="Identifier of target TON account in raw form.", + example="0:83DFD552E63729B472FCBCC8C45EBCC6691702558B68EC7527E1BA403A0F31A8", + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Convert an address from raw to human-readable format. """ return prepare_address(address) -@router.get('/unpackAddress', response_model=TonResponse, response_model_exclude_none=True, tags=['accounts']) -@json_rpc('unpackAddress') + +@router.get( + "/unpackAddress", + response_model=UnpackAddressResponse, + responses=get_get_address_information_error_responses(), + response_model_exclude_none=True, + tags=["accounts"], +) +@json_rpc("unpackAddress") @wrap_result async def unpack_address( - address: str = Query(..., description="Identifier of target TON account in user-friendly form", example="EQCD39VS5jcptHL8vMjEXrzGaRcCVYto7HUn4bpAOg8xqB2N") - ): + address: str = Query( + ..., + description="Identifier of target TON account in user-friendly form", + example="EQCD39VS5jcptHL8vMjEXrzGaRcCVYto7HUn4bpAOg8xqB2N", + ) +): """ Convert an address from human-readable to raw format. """ return _detect_address(address)["raw_form"] -@router.get('/getMasterchainInfo', response_model=TonResponse, response_model_exclude_none=True, tags=['blocks']) -@json_rpc('getMasterchainInfo') + +@router.get( + "/getMasterchainInfo", + response_model=TonResponse200Generic[MasterchainInfo], + response_model_exclude_none=True, + tags=["blocks"], +) +@json_rpc("getMasterchainInfo") @wrap_result -async def get_masterchain_info(tonlib: TonlibManager=Depends(tonlib_dep)): +async def get_masterchain_info(tonlib: TonlibManager = Depends(tonlib_dep)): """ Get up-to-date masterchain state. """ return await tonlib.getMasterchainInfo() -@router.get('/getMasterchainBlockSignatures', response_model=TonResponse, response_model_exclude_none=True, tags=['blocks']) -@json_rpc('getMasterchainBlockSignatures') + +@router.get( + "/getMasterchainBlockSignatures", + response_model=TonResponse200Generic[MasterchainSignatures], + response_model_exclude_none=True, + tags=["blocks"], +) +@json_rpc("getMasterchainBlockSignatures") @wrap_result async def get_masterchain_block_signatures( - seqno: int, - tonlib: TonlibManager=Depends(tonlib_dep) - ): + seqno: int, tonlib: TonlibManager = Depends(tonlib_dep) +): """ Get up-to-date masterchain state. """ return await tonlib.getMasterchainBlockSignatures(seqno) -@router.get('/getShardBlockProof', response_model=TonResponse, response_model_exclude_none=True, tags=['blocks']) -@json_rpc('getShardBlockProof') + +@router.get( + "/getShardBlockProof", + response_model=TonResponse200Generic[ShardBlockProof], + response_model_exclude_none=True, + tags=["blocks"], +) +@json_rpc("getShardBlockProof") @wrap_result async def get_shard_block_proof( workchain: int = Query(..., description="Block workchain id"), - shard: int = Query(..., description="Block shard id"), + shard: int = Query(..., description="Block shard id"), seqno: int = Query(..., description="Block seqno"), - from_seqno: Optional[int] = Query(None, description="Seqno of masterchain block starting from which proof is required. If not specified latest masterchain block is used."), - tonlib: TonlibManager=Depends(tonlib_dep) - ): + from_seqno: Optional[int] = Query( + None, + description="Seqno of masterchain block starting from which proof is required. If not specified latest masterchain block is used.", + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Get merkle proof of shardchain block. """ return await tonlib.getShardBlockProof(workchain, shard, seqno, from_seqno) -@router.get('/getConsensusBlock', response_model=TonResponse, response_model_exclude_none=True, tags=['blocks']) -@json_rpc('getConsensusBlock') +@router.get( + "/getConsensusBlock", + response_model=TonResponse200Generic[ConsensusBlock], + response_model_exclude_none=True, + tags=["blocks"], +) +@json_rpc("getConsensusBlock") @wrap_result -async def get_consensus_block(tonlib: TonlibManager=Depends(tonlib_dep)): +async def get_consensus_block(tonlib: TonlibManager = Depends(tonlib_dep)): """ Get consensus block and its update timestamp. """ return await tonlib.getConsensusBlock() -@router.get('/lookupBlock', response_model=TonResponse, response_model_exclude_none=True, tags=['blocks']) -@json_rpc('lookupBlock') + +@router.get( + "/lookupBlock", + response_model=TonResponse200Generic[BlockId], + response_model_exclude_none=True, + tags=["blocks"], +) +@json_rpc("lookupBlock") @wrap_result async def lookup_block( - workchain: int = Query(..., description="Workchain id to look up block in"), + workchain: int = Query(..., description="Workchain id to look up block in"), shard: int = Query(..., description="Shard id to look up block in"), seqno: Optional[int] = Query(None, description="Block's height"), - lt: Optional[int] = Query(None, description="Block's logical time"), + lt: Optional[int] = Query(None, description="Block's logical time"), unixtime: Optional[int] = Query(None, description="Block's unixtime"), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Look up block by either *seqno*, *lt* or *unixtime*. """ return await tonlib.lookupBlock(workchain, shard, seqno, lt, unixtime) -@router.get('/shards', response_model=TonResponse, response_model_exclude_none=True, tags=['blocks']) -@json_rpc('shards') + +@router.get( + "/shards", + response_model=TonResponse200Generic[Shards], + response_model_exclude_none=True, + tags=["blocks"], +) +@json_rpc("shards") @wrap_result async def get_shards( seqno: int = Query(..., description="Masterchain seqno to fetch shards of."), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Get shards information. """ return await tonlib.getShards(seqno) -@router.get('/getBlockTransactions', response_model=TonResponse, response_model_exclude_none=True, tags=['blocks','transactions']) -@json_rpc('getBlockTransactions') + +@router.get( + "/getBlockTransactions", + response_model=TonResponse200Generic[ShortTransactions], + response_model_exclude_none=True, + tags=["blocks", "transactions"], +) +@json_rpc("getBlockTransactions") @wrap_result async def get_block_transactions( - workchain: int, - shard: int, - seqno: int, - root_hash: Optional[str] = None, - file_hash: Optional[str] = None, - after_lt: Optional[int] = None, - after_hash: Optional[str] = None, + workchain: int, + shard: int, + seqno: int, + root_hash: Optional[str] = None, + file_hash: Optional[str] = None, + after_lt: Optional[int] = None, + after_hash: Optional[str] = None, count: int = 40, - tonlib: TonlibManager = Depends(tonlib_dep) - ): + tonlib: TonlibManager = Depends(tonlib_dep), +): """ - Get transactions of the given block. + Get transactions metadata of the given block. """ - return await tonlib.getBlockTransactions(workchain, shard, seqno, count, root_hash, file_hash, after_lt, after_hash) + return await tonlib.getBlockTransactions( + workchain, shard, seqno, count, root_hash, file_hash, after_lt, after_hash + ) + -@router.get('/getBlockHeader', response_model=TonResponse, response_model_exclude_none=True, tags=['blocks']) -@json_rpc('getBlockHeader') +@router.get( + "/getBlockHeader", + response_model=TonResponse200Generic[BlockHeader], + response_model_exclude_none=True, + tags=["blocks"], +) +@json_rpc("getBlockHeader") @wrap_result async def get_block_header( - workchain: int, - shard: int, - seqno: int, - root_hash: Optional[str] = None, + workchain: int, + shard: int, + seqno: int, + root_hash: Optional[str] = None, file_hash: Optional[str] = None, - tonlib: TonlibManager = Depends(tonlib_dep) - ): + tonlib: TonlibManager = Depends(tonlib_dep), +): """ - Get metadata of a given block. + Get metadata of the given block. """ return await tonlib.getBlockHeader(workchain, shard, seqno, root_hash, file_hash) -@router.get('/getConfigParam', response_model=TonResponse, response_model_exclude_none=True, tags=['get config']) -@json_rpc('getConfigParam') + +@router.get( + "/getConfigParam", + response_model=GetConfigParamResponse, + response_model_exclude_none=True, + tags=["get config"], +) +@json_rpc("getConfigParam") @wrap_result async def get_config_param( config_id: int = Query(..., description="Config id"), - seqno: Optional[int] = Query(None, description="Masterchain seqno. If not specified, latest blockchain state will be used."), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + seqno: Optional[int] = Query( + None, + description="Masterchain seqno. If not specified, latest blockchain state will be used.", + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Get config by id. + Information about cell content can be found [in TL-B definitions](https://github.com/ton-blockchain/ton/blob/5c392e0f2d946877bb79a09ed35068f7b0bd333a/crypto/block/block.tlb#L593). """ return await tonlib.get_config_param(config_id, seqno) -@router.get('/getTokenData', response_model=TonResponse, response_model_exclude_none=True, tags=['accounts']) -@json_rpc('getTokenData') + +@router.get( + "/getTokenData", + response_model=TonGetTokenDataResponse, + responses=get_get_address_information_error_responses(), + response_model_exclude_none=True, + tags=["accounts"], +) +@json_rpc("getTokenData") @wrap_result async def get_token_data( - address: str = Query(..., description="Address of NFT collection/item or Jetton master/wallet smart contract"), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Query( + ..., + description="Address of NFT collection/item or Jetton master/wallet smart contract", + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Get NFT or Jetton information. """ address = prepare_address(address) return await tonlib.get_token_data(address) -@router.get('/tryLocateTx', response_model=TonResponse, response_model_exclude_none=True, tags=['transactions']) -@json_rpc('tryLocateTx') + +@router.get( + "/tryLocateTx", + response_model=TonResponse200Generic[RawTransaction[TransactionId]], + response_model_exclude_none=True, + tags=["transactions"], +) +@json_rpc("tryLocateTx") @wrap_result async def get_try_locate_tx( - source: str, - destination: str, + source: str, + destination: str, created_lt: int, - tonlib: TonlibManager = Depends(tonlib_dep) - ): + tonlib: TonlibManager = Depends(tonlib_dep), +): """ - Locate outcoming transaction of *destination* address by incoming message. + Locate transaction for destination address by incoming message. """ return await tonlib.tryLocateTxByIncomingMessage(source, destination, created_lt) -@router.get('/tryLocateResultTx', response_model=TonResponse, response_model_exclude_none=True, tags=['transactions']) -@json_rpc('tryLocateResultTx') + +@router.get( + "/tryLocateResultTx", + response_model=TonResponse200Generic[RawTransaction[TransactionId]], + response_model_exclude_none=True, + tags=["transactions"], +) +@json_rpc("tryLocateResultTx") @wrap_result async def get_try_locate_result_tx( - source: str, - destination: str, + source: str, + destination: str, created_lt: int, - tonlib: TonlibManager = Depends(tonlib_dep) - ): + tonlib: TonlibManager = Depends(tonlib_dep), +): """ - Same as previous. Locate outcoming transaction of *destination* address by incoming message + Same as previous. Locate transaction for destination address by incoming message. """ return await tonlib.tryLocateTxByIncomingMessage(source, destination, created_lt) -@router.get('/tryLocateSourceTx', response_model=TonResponse, response_model_exclude_none=True, tags=['transactions']) -@json_rpc('tryLocateSourceTx') + +@router.get( + "/tryLocateSourceTx", + response_model=TonResponse200Generic[RawTransaction[TransactionId]], + response_model_exclude_none=True, + tags=["transactions"], +) +@json_rpc("tryLocateSourceTx") @wrap_result async def get_try_locate_source_tx( - source: str, - destination: str, + source: str, + destination: str, created_lt: int, - tonlib: TonlibManager = Depends(tonlib_dep) - ): + tonlib: TonlibManager = Depends(tonlib_dep), +): """ - Locate incoming transaction of *source* address by outcoming message. + Locate transaction for source address by outcoming message. """ return await tonlib.tryLocateTxByOutcomingMessage(source, destination, created_lt) -@router.get('/detectAddress', response_model=TonResponse, response_model_exclude_none=True, tags=['accounts']) -@json_rpc('detectAddress') + +@router.get( + "/detectAddress", + response_model=TonResponse200Generic[AddressForms], + responses=get_get_address_information_error_responses(), + response_model_exclude_none=True, + tags=["accounts"], +) +@json_rpc("detectAddress") @wrap_result async def detect_address( - address: str = Query(..., description="Identifier of target TON account in any form.") - ): + address: str = Query( + ..., description="Identifier of target TON account in any form." + ) +): """ Get all possible address forms. """ @@ -422,202 +668,310 @@ def send_boc_to_external_endpoint(boc): try: endpoint = settings.webserver.boc_endpoint logger.info(f'BOC is: "{boc}"') - res = requests.post(endpoint, json={'boc': boc}) + res = requests.post(endpoint, json={"boc": boc}) logger.info(f"Boc sent to external endpoint: {res}") - return res.get('ok', False) + return res.get("ok", False) except Exception as ee: - logger.warning(f'Failed to sent a message to external endpoint: {ee}') + logger.warning(f"Failed to sent a message to external endpoint: {ee}") except: - logger.warning(f'Failed to sent a message to external endpoint: unknown') + logger.warning(f"Failed to sent a message to external endpoint: unknown") return False -@router.post('/sendBoc', response_model=TonResponse, response_model_exclude_none=True, tags=['send']) -@json_rpc('sendBoc') +@router.post( + "/sendBoc", + response_model=OkResponse, + response_model_exclude_none=True, + tags=["send"], +) +@json_rpc("sendBoc") @wrap_result async def send_boc( background_tasks: BackgroundTasks, boc: str = Body(..., embed=True, description="b64 encoded bag of cells"), tonlib: TonlibManager = Depends(tonlib_dep), - ): +): """ Send serialized boc file: fully packed and serialized external message to blockchain. """ boc = base64.b64decode(boc) res = await tonlib.raw_send_message(boc) - if res.get('@type') == 'ok': + if res.get("@type") == "ok": logger.debug("External message accepted in sendBoc") if settings.webserver.boc_endpoint is not None: - background_tasks.add_task(send_boc_to_external_endpoint, base64.b64encode(boc).decode('utf8')) + background_tasks.add_task( + send_boc_to_external_endpoint, base64.b64encode(boc).decode("utf8") + ) return res -@router.post('/sendBocReturnHash', response_model=TonResponse, response_model_exclude_none=True, tags=['send']) -@json_rpc('sendBocReturnHash') + +@router.post( + "/sendBocReturnHash", + response_model=SendBocReturnHashResponse, + response_model_exclude_none=True, + tags=["send"], +) +@json_rpc("sendBocReturnHash") @wrap_result async def send_boc_return_hash( background_tasks: BackgroundTasks, boc: str = Body(..., embed=True, description="b64 encoded bag of cells"), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Send serialized boc file: fully packed and serialized external message to blockchain. The method returns message hash. """ boc = base64.b64decode(boc) res = await tonlib.raw_send_message_return_hash(boc) - if res.get('@type') == 'raw.extMessageInfo': - logger.info("External message accepted: {hash}", hash=res.get('hash')) + if res.get("@type") == "raw.extMessageInfo": + logger.info("External message accepted: {hash}", hash=res.get("hash")) if settings.webserver.boc_endpoint is not None: - background_tasks.add_task(send_boc_to_external_endpoint, base64.b64encode(boc).decode('utf8')) + background_tasks.add_task( + send_boc_to_external_endpoint, base64.b64encode(boc).decode("utf8") + ) return res + async def send_boc_unsafe_task(boc_bytes: bytes, tonlib: TonlibManager): send_interval = 5 send_duration = 60 for i in range(int(send_duration / send_interval)): try: res = await tonlib.raw_send_message(boc_bytes) - if res.get('@type') == 'ok': - logger.debug('External message accepted in sendBocUnsafe') + if res.get("@type") == "ok": + logger.debug("External message accepted in sendBocUnsafe") if settings.webserver.boc_endpoint is not None: - send_boc_to_external_endpoint(base64.b64encode(boc_bytes).decode('utf8')) + send_boc_to_external_endpoint( + base64.b64encode(boc_bytes).decode("utf8") + ) except: pass await asyncio.sleep(send_interval) -@router.post('/sendBocUnsafe', response_model=TonResponse, response_model_exclude_none=True, include_in_schema=False, tags=['send']) -@json_rpc('sendBocUnsafe') + +@router.post( + "/sendBocUnsafe", + response_model=TonResponse, + response_model_exclude_none=True, + include_in_schema=False, + tags=["send"], +) +@json_rpc("sendBocUnsafe") @wrap_result async def send_boc_unsafe( background_tasks: BackgroundTasks, boc: str = Body(..., embed=True, description="b64 encoded bag of cells"), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Unsafe send serialized boc file: fully packed and serialized external message to blockchain. This method creates background task that sends boc to network every 5 seconds for 1 minute. """ boc = base64.b64decode(boc) background_tasks.add_task(send_boc_unsafe_task, boc, tonlib) - return {'@type': 'ok', '@extra': '0:0:0'} + return {"@type": "ok", "@extra": "0:0:0"} + -@router.post('/sendCellSimple', response_model=TonResponse, response_model_exclude_none=True, include_in_schema=False, tags=['send']) -@json_rpc('sendCellSimple') +@router.post( + "/sendCellSimple", + response_model=TonResponse, + response_model_exclude_none=True, + include_in_schema=False, + tags=["send"], +) +@json_rpc("sendCellSimple") @wrap_result async def send_cell( - cell: Dict[str, Any] = Body(..., embed=True, description="Cell serialized as object"), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + cell: Dict[str, Any] = Body( + ..., embed=True, description="Cell serialized as object" + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ (Deprecated) Send cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`, that is fully packed but not serialized external message. """ try: cell = deserialize_cell_from_object(cell) - boc = codecs.encode(cell.serialize_boc(), 'base64') + boc = codecs.encode(cell.serialize_boc(), "base64") except: raise HTTPException(status_code=400, detail="Error while parsing cell") return await tonlib.raw_send_message(boc) -@router.post('/sendQuery', response_model=TonResponse, response_model_exclude_none=True, tags=['send']) -@json_rpc('sendQuery') + +@router.post( + "/sendQuery", + response_model=OkResponse, + response_model_exclude_none=True, + tags=["send"], +) +@json_rpc("sendQuery") @wrap_result async def send_query( - address: str = Body(..., description="Address in any format"), - body: str = Body(..., description="b64-encoded boc-serialized cell with message body"), - init_code: str = Body(default='', description="b64-encoded boc-serialized cell with init-code"), - init_data: str = Body(default='', description="b64-encoded boc-serialized cell with init-data"), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Body(..., description="Address in any format"), + body: str = Body( + ..., + description="b64-encoded boc-serialized cell with message body", + ), + init_code: str = Body( + default="", description="b64-encoded boc-serialized cell with init-code" + ), + init_data: str = Body( + default="", description="b64-encoded boc-serialized cell with init-data" + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Send query - unpacked external message. This method takes address, body and init-params (if any), packs it to external message and sends to network. All params should be boc-serialized. """ address = prepare_address(address) - body = codecs.decode(codecs.encode(body, "utf-8"), 'base64') - code = codecs.decode(codecs.encode(init_code, "utf-8"), 'base64') - data = codecs.decode(codecs.encode(init_data, "utf-8"), 'base64') - return await tonlib.raw_create_and_send_query(address, body, init_code=code, init_data=data) - -@router.post('/sendQuerySimple', response_model=TonResponse, response_model_exclude_none=True, include_in_schema=False, tags=['send']) -@json_rpc('sendQuerySimple') + body = codecs.decode(codecs.encode(body, "utf-8"), "base64") + code = codecs.decode(codecs.encode(init_code, "utf-8"), "base64") + data = codecs.decode(codecs.encode(init_data, "utf-8"), "base64") + return await tonlib.raw_create_and_send_query( + address, body, init_code=code, init_data=data + ) + + +@router.post( + "/sendQuerySimple", + response_model=TonResponse, + response_model_exclude_none=True, + include_in_schema=False, + tags=["send"], +) +@json_rpc("sendQuerySimple") @wrap_result async def send_query_cell( - address: str = Body(..., description="Address in any format"), - body: str = Body(..., description='Body cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`'), - init_code: Optional[Dict[str, Any]] = Body(default=None, description='init-code cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`'), - init_data: Optional[Dict[str, Any]] = Body(default=None, description='init-data cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`'), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Body(..., description="Address in any format"), + body: str = Body( + ..., + description='Body cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`', + ), + init_code: Optional[Dict[str, Any]] = Body( + default=None, + description='init-code cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`', + ), + init_data: Optional[Dict[str, Any]] = Body( + default=None, + description='init-data cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`', + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ (Deprecated) Send query - unpacked external message. This method gets address, body and init-params (if any), packs it to external message and sends to network. Body, init-code and init-data should be passed as objects. """ address = prepare_address(address) try: body = deserialize_cell_from_object(body).serialize_boc(has_idx=False) - qcode, qdata = b'', b'' + qcode, qdata = b"", b"" if init_code is not None: qcode = deserialize_cell_from_object(init_code).serialize_boc(has_idx=False) if init_data is not None: qdata = deserialize_cell_from_object(init_data).serialize_boc(has_idx=False) except: raise HTTPException(status_code=400, detail="Error while parsing cell object") - return await tonlib.raw_create_and_send_query(address, body, init_code=qcode, init_data=qdata) + return await tonlib.raw_create_and_send_query( + address, body, init_code=qcode, init_data=qdata + ) + -@router.post('/estimateFee', response_model=TonResponse, response_model_exclude_none=True, tags=['send']) -@json_rpc('estimateFee') +@router.post( + "/estimateFee", + response_model=EstimateFeeResponse, + response_model_exclude_none=True, + tags=["send"], +) +@json_rpc("estimateFee") @wrap_result async def estimate_fee( - address: str = Body(..., description='Address in any format'), - body: str = Body(..., description='b64-encoded cell with message body'), - init_code: str = Body(default='', description='b64-encoded cell with init-code'), - init_data: str = Body(default='', description='b64-encoded cell with init-data'), - ignore_chksig: bool = Body(default=True, description='If true during test query processing assume that all chksig operations return True'), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Body(..., description="Address in any format"), + body: str = Body(..., description="b64-encoded cell with message body"), + init_code: str = Body(default="", description="b64-encoded cell with init-code"), + init_data: str = Body(default="", description="b64-encoded cell with init-data"), + ignore_chksig: bool = Body( + default=True, + description="If true during test query processing assume that all chksig operations return True", + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ Estimate fees required for query processing. *body*, *init-code* and *init-data* accepted in serialized format (b64-encoded). """ address = prepare_address(address) - body = codecs.decode(codecs.encode(body, "utf-8"), 'base64') - code = codecs.decode(codecs.encode(init_code, "utf-8"), 'base64') - data = codecs.decode(codecs.encode(init_data, "utf-8"), 'base64') - return await tonlib.raw_estimate_fees(address, body, init_code=code, init_data=data, ignore_chksig=ignore_chksig) - -@router.post('/estimateFeeSimple', response_model=TonResponse, response_model_exclude_none=True, include_in_schema=False, tags=['send']) -@json_rpc('estimateFeeSimple') + body = codecs.decode(codecs.encode(body, "utf-8"), "base64") + code = codecs.decode(codecs.encode(init_code, "utf-8"), "base64") + data = codecs.decode(codecs.encode(init_data, "utf-8"), "base64") + return await tonlib.raw_estimate_fees( + address, body, init_code=code, init_data=data, ignore_chksig=ignore_chksig + ) + + +@router.post( + "/estimateFeeSimple", + response_model=TonResponse, + response_model_exclude_none=True, + include_in_schema=False, + tags=["send"], +) +@json_rpc("estimateFeeSimple") @wrap_result async def estimate_fee_cell( - address: str = Body(..., description='Address in any format'), - body: Dict[str, Any] = Body(..., description='Body cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`'), - init_code: Optional[Dict[str, Any]] = Body(default=None, description='init-code cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`'), - init_data: Optional[Dict[str, Any]] = Body(default=None, description='init-data cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`'), - ignore_chksig: bool = Body(default=True, description='If true during test query processing assume that all chksig operations return True'), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Body(..., description="Address in any format"), + body: Dict[str, Any] = Body( + ..., + description='Body cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`', + ), + init_code: Optional[Dict[str, Any]] = Body( + default=None, + description='init-code cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`', + ), + init_data: Optional[Dict[str, Any]] = Body( + default=None, + description='init-data cell as object: `{"data": {"b64": "...", "len": int }, "refs": [...subcells...]}`', + ), + ignore_chksig: bool = Body( + default=True, + description="If true during test query processing assume that all chksig operations return True", + ), + tonlib: TonlibManager = Depends(tonlib_dep), +): """ (Deprecated) Estimate fees required for query processing. *body*, *init-code* and *init-data* accepted in unserialized format (as objects). """ address = prepare_address(address) try: body = deserialize_cell_from_object(body).serialize_boc(has_idx=False) - qcode, qdata = b'', b'' + qcode, qdata = b"", b"" if init_code is not None: qcode = deserialize_cell_from_object(init_code).serialize_boc(has_idx=False) if init_data is not None: qdata = deserialize_cell_from_object(init_data).serialize_boc(has_idx=False) except: raise HTTPException(status_code=400, detail="Error while parsing cell object") - return await tonlib.raw_estimate_fees(address, body, init_code=qcode, init_data=qdata, ignore_chksig=ignore_chksig) + return await tonlib.raw_estimate_fees( + address, body, init_code=qcode, init_data=qdata, ignore_chksig=ignore_chksig + ) if settings.webserver.get_methods: - @router.post('/runGetMethod', response_model=TonResponse, response_model_exclude_none=True, tags=["run method"]) - @json_rpc('runGetMethod') + + @router.post( + "/runGetMethod", + response_model=RunGetMethodResponse, + response_model_exclude_none=True, + tags=["run method"], + ) + @json_rpc("runGetMethod") @wrap_result async def run_get_method( - address: str = Body(..., description='Contract address'), - method: Union[str, int] = Body(..., description='Method name or method id'), - stack: List[List[Any]] = Body(..., description="Array of stack elements: `[['num',3], ['cell', cell_object], ['slice', slice_object]]`"), - tonlib: TonlibManager = Depends(tonlib_dep) - ): + address: str = Body(..., description="Contract address"), + method: Union[str, int] = Body(..., description="Method name or method id"), + stack: List[List[Any]] = Body( + ..., + description="Array of stack elements: `[['num',3], ['cell', cell_object], ['slice', slice_object]]`", + ), + tonlib: TonlibManager = Depends(tonlib_dep), + ): """ Run get method on smart contract. """ @@ -626,14 +980,22 @@ async def run_get_method( if settings.webserver.json_rpc: - @router.post('/jsonRPC', response_model=DeprecatedTonResponseJsonRPC, response_model_exclude_none=True, tags=['json rpc']) - async def jsonrpc_handler(json_rpc: TonRequestJsonRPC, - request: Request, - response: Response, - background_tasks: BackgroundTasks, - tonlib: TonlibManager = Depends(tonlib_dep)): + + @router.post( + "/jsonRPC", + response_model=DeprecatedTonResponseJsonRPC, + response_model_exclude_none=True, + tags=["json rpc"], + ) + async def jsonrpc_handler( + json_rpc: TonRequestJsonRPC, + request: Request, + response: Response, + background_tasks: BackgroundTasks, + tonlib: TonlibManager = Depends(tonlib_dep), + ): """ - All methods in the API are available through JSON-RPC protocol ([spec](https://www.jsonrpc.org/specification)). + All methods in the API are available through JSON-RPC protocol ([spec](https://www.jsonrpc.org/specification)). """ params = json_rpc.params method = json_rpc.method @@ -641,20 +1003,30 @@ async def jsonrpc_handler(json_rpc: TonRequestJsonRPC, if not method in json_rpc_methods: response.status_code = status.HTTP_422_UNPROCESSABLE_ENTITY - return DeprecatedTonResponseJsonRPC(ok=False, error='Unknown method', id=_id) + return DeprecatedTonResponseJsonRPC( + ok=False, error="Unknown method", id=_id + ) handler = json_rpc_methods[method] try: - if 'request' in inspect.signature(handler).parameters.keys(): - params['request'] = request - if 'background_tasks' in inspect.signature(handler).parameters.keys(): - params['background_tasks'] = background_tasks - if 'tonlib' in inspect.signature(handler).parameters.keys(): - params['tonlib'] = tonlib + if "request" in inspect.signature(handler).parameters.keys(): + params["request"] = request + if "background_tasks" in inspect.signature(handler).parameters.keys(): + params["background_tasks"] = background_tasks + if "tonlib" in inspect.signature(handler).parameters.keys(): + params["tonlib"] = tonlib result = await handler(**params) except TypeError as e: response.status_code = status.HTTP_422_UNPROCESSABLE_ENTITY - return DeprecatedTonResponseJsonRPC(ok=False, error=f'TypeError: {e}', id=_id) - - return DeprecatedTonResponseJsonRPC(ok=result.ok, result=result.result, error=result.error, code=result.code, id=_id) + return DeprecatedTonResponseJsonRPC( + ok=False, error=f"TypeError: {e}", id=_id + ) + + return DeprecatedTonResponseJsonRPC( + ok=result.ok, + result=result.result, + error=result.error, + code=result.code, + id=_id, + ) diff --git a/ton-http-api/pyTON/api/api_v3/endpoints/common.py b/ton-http-api/pyTON/api/api_v3/endpoints/common.py index d83a3eb..83d21ed 100644 --- a/ton-http-api/pyTON/api/api_v3/endpoints/common.py +++ b/ton-http-api/pyTON/api/api_v3/endpoints/common.py @@ -613,7 +613,8 @@ async def jsonrpc_handler(json_rpc: TonRequestJsonRPC, background_tasks: BackgroundTasks, tonlib: TonlibManager = Depends(tonlib_dep)): """ - All methods in the API are available through JSON-RPC protocol ([spec](https://www.jsonrpc.org/specification)). + All methods in the API are available through JSON-RPC protocol ([spec](https://www.jsonrpc.org/specification)). + The type of the "result" field is the same as `result` field in original method. """ params = json_rpc.params method = json_rpc.method diff --git a/ton-http-api/pyTON/schemas/__init__.py b/ton-http-api/pyTON/schemas/__init__.py index ae75d85..a94b670 100644 --- a/ton-http-api/pyTON/schemas/__init__.py +++ b/ton-http-api/pyTON/schemas/__init__.py @@ -1,9 +1,24 @@ from .http import ( TonRequestJsonRPC, TonResponse, - TonResponseGeneric, TonResponseJsonRPC, - DeprecatedTonResponseJsonRPC + DeprecatedTonResponseJsonRPC, + get_get_address_information_error_responses, + GetAddressInformationResponse, + GetExtendedAddressInformationResponse, + GetWalletInformationResponse, + GetAddressBalanceResponse, + GetAddressStateResponse, + PackAddressResponse, + UnpackAddressResponse, + TonGetTokenDataResponse, + TonResponse200Generic, + GetConfigParamResponse, + RunGetMethodResponse, + DeprecatedTonResponseJsonRPC, + OkResponse, + SendBocReturnHashResponse, + EstimateFeeResponse, ) from .ton import ( BlockId, @@ -18,6 +33,13 @@ Message, TransactionId, Transaction, + MasterchainSignatures, + ShardBlockProof, + ConsensusBlock, + Shards, + ShortTransactions, + ShortTransactions, + TransactionWAddressId, check_tonlib_type, - address_state + address_state, ) diff --git a/ton-http-api/pyTON/schemas/http.py b/ton-http-api/pyTON/schemas/http.py index 13b5e75..8f0aa44 100644 --- a/ton-http-api/pyTON/schemas/http.py +++ b/ton-http-api/pyTON/schemas/http.py @@ -1,41 +1,238 @@ -from typing import Optional, TypeVar, Union +from typing import Optional, TypeVar, Union, List, Literal, Tuple, Any from pydantic.generics import GenericModel, Generic -from pydantic import BaseModel +from pydantic import BaseModel, Field +from .ton import ( + AccountStateRow, + AccountStateUninited, + AccountStateWallet, + ConfigInfo, + TVMStackEntryType, + TvmTuple, + BlockIdExt, + TransactionId, + BlockId, + AddressShort, + JettonMasterData, + JettonWalletData, + NftCollectionData, + NftItemData, +) -ResultT = TypeVar('ResultT') +ResultT = TypeVar("ResultT") +ResultT2 = TypeVar("ResultT2") class TonResponseGeneric(GenericModel, Generic[ResultT]): ok: bool - result: Optional[ResultT] + result: Optional[Union[ResultT, Any]] error: Optional[str] = None code: Optional[int] = None +class TonResponse200Generic(GenericModel, Generic[ResultT2]): + ok: bool = Field(True) + result: Optional[Union[ResultT2, Any]] + + +ResultTypeT = TypeVar("ResultTypeT") + + +class TonResponseResultGeneric(GenericModel, Generic[ResultTypeT]): + type: ResultTypeT = Field(alias="@type") + extra: str = Field(alias="@extra") + + class TonResponse(TonResponseGeneric[Union[str, list, dict, None]]): pass +class ErrorGetAddressInformationResponses422(BaseModel): + ok: bool = Field(False) + error: str + code: int = Field(422) + + @staticmethod + def get_response(): + return { + 422: { + "model": ErrorGetAddressInformationResponses422, + "description": "Validation Error", + } + } + + +class ErrorGetAddressInformationResponses504(BaseModel): + ok: bool = Field(False) + error: str + code: int = Field(504) + + @staticmethod + def get_response(): + return { + 504: { + "model": ErrorGetAddressInformationResponses504, + "description": "Lite Server Timeout", + } + } + + +def get_get_address_information_error_responses(): + response = ErrorGetAddressInformationResponses422.get_response() + response.update(ErrorGetAddressInformationResponses504.get_response()) + return response + + +class GetAddressInformationResponse(BaseModel): + type: str = Field(alias="@type") + balance: str + code: str + data: str + last_transaction_id: TransactionId + block_id: BlockId + frozen_hash: str + sync_utime: int + extra: str = Field(alias="@extra") + state: str + + +class GetExtendedAddressInformationResponse(BaseModel): + type: str = Field(alias="@type") + address: AddressShort + balance: str + last_transaction_id: TransactionId + block_id: BlockId + sync_utime: int + account_state: Union[AccountStateWallet, AccountStateRow, AccountStateUninited, Any] + revision: int + extra: str = Field(alias="@extra") + + +class GetWalletInformationResponse(BaseModel): + wallet: bool + balance: str + account_state: str + wallet_type: str + seqno: Optional[int] + last_transaction_id: TransactionId + wallet_id: Optional[int] + + +class GetAddressBalanceResponse(BaseModel): + ok: bool = Field(example=True) + result: str = Field( + example="1234", + description="str representation of number, balance of the contract", + ) + + +class GetAddressStateResponse(BaseModel): + ok: bool = Field(True) + result: Literal["nonexist", "uninit", "active", "frozen"] = Field( + description="State of the address, visit https://docs.ton.org/learn/overviews/addresses#addresses-state for more" + ) + + +class PackAddressResponse(BaseModel): + ok: bool = Field(True) + result: str = Field( + example="EQCD39VS5jcptHL8vMjEXrzGaRcCVYto7HUn4bpAOg8xqB2N", + description="Packed address", + ) + + +class UnpackAddressResponse(BaseModel): + ok: bool = Field(True) + result: str = Field( + example="0:83dfd552e63729b472fcbcc8c45ebcc6691702558b68ec7527e1ba403a0f31a8", + description="Unpacked address", + ) + + +class TonGetTokenDataResponse(BaseModel): + ok: bool = Field(True) + result: Union[ + JettonMasterData, NftCollectionData, NftItemData, JettonWalletData, Any + ] + + class TonResponseJsonRPC(BaseModel): id: str jsonrpc: str = "2.0" - result: Optional[ResultT] + result: Optional[Union[ResultT, Any]] error: Optional[str] = None code: Optional[int] = None class DeprecatedTonResponseJsonRPC(BaseModel): ok: bool - result: Optional[ResultT] + result: Optional[Union[ResultT, Any]] error: Optional[str] = None code: Optional[int] = None id: str - jsonrpc: str = "2.0" + jsonrpc: Literal["2.0"] = "2.0" class TonRequestJsonRPC(BaseModel): - method: str - params: dict = {} + method: str = Field(example="runGetMethod") + params: dict = Field( + {}, + example={ + "address": "kQAl8r8c6Pg-0MD9c-onqsdwk83PkAx1Cwcd9_sCiOAZsoUE", + "method": "get_jetton_data", + "stack": [], + }, + ) id: Optional[str] = None jsonrpc: Optional[str] = None + + +class GetConfigParamResponse(TonResponseGeneric[ConfigInfo]): + pass + + +class RunGetMethodResult(BaseModel): + type: Literal["smc.runResult"] = Field(alias="@type") + gas_used: int + stack: List[List[Union[TVMStackEntryType, Union[str, TvmTuple]]]] = Field( + example=[["num", "0x1"]] + ) + exit_code: int + extra: str = Field(alias="@extra") + block_id: BlockIdExt + last_transaction_id: TransactionId + + +class RunGetMethodResponse(TonResponseGeneric[RunGetMethodResult]): + pass + + +class OkResponse(TonResponseGeneric[TonResponseResultGeneric[Literal["ok"]]]): + pass + + +class SendBocReturnHashResult(TonResponseResultGeneric[Literal["raw.extMessageInfo"]]): + hash: str = Field(example="65+BlkfroywqXyM+POVpMpFiC6XYMQyBvHXw12XiFzc=") + + +class SendBocReturnHashResponse(TonResponseGeneric[SendBocReturnHashResult]): + pass + + +class Fees(BaseModel): + type: Literal["fees"] = Field(alias="@type") + in_fwd_fee: int + storage_fee: int + gas_fee: int + fwd_fee: int + + +class EstimateFeeResponseResult(BaseModel): + type: Literal["query.fees"] = Field(alias="@type") + source_fees: Fees + destination_fees: List[Fees] + extra: str = Field(alias="@extra") + + +class EstimateFeeResponse(TonResponseGeneric[EstimateFeeResponseResult]): + pass diff --git a/ton-http-api/pyTON/schemas/ton.py b/ton-http-api/pyTON/schemas/ton.py index f63cb23..65c1ece 100644 --- a/ton-http-api/pyTON/schemas/ton.py +++ b/ton-http-api/pyTON/schemas/ton.py @@ -1,12 +1,14 @@ - -from typing import List, Optional, Literal -from pydantic import BaseModel +from typing import List, Optional, Literal, TypeVar, Union +from pydantic import BaseModel, Field +from pydantic.generics import GenericModel, Generic from pytonlib.utils.wallet import wallets as known_wallets, sha256 +ResultT = TypeVar("ResultT") + def check_tonlib_type(tl_obj: dict, expected_type: str): - tl_type = tl_obj.get('@type', '') + tl_type = tl_obj.get("@type", "") if tl_type != expected_type: raise Exception(f"Unexpected TL object type {tl_type}") @@ -28,18 +30,25 @@ class BlockId(BaseModel): file_hash: str def build(tl_obj: dict): - check_tonlib_type(tl_obj, 'ton.blockIdExt') - - workchain = int(tl_obj['workchain']) - shard = tl_obj['shard'] - seqno = int(tl_obj['seqno']) - root_hash = tl_obj['root_hash'] - file_hash = tl_obj['file_hash'] - - return BlockId(workchain=workchain, shard=shard, seqno=seqno, root_hash=root_hash, file_hash=file_hash) + check_tonlib_type(tl_obj, "ton.blockIdExt") + + workchain = int(tl_obj["workchain"]) + shard = tl_obj["shard"] + seqno = int(tl_obj["seqno"]) + root_hash = tl_obj["root_hash"] + file_hash = tl_obj["file_hash"] + + return BlockId( + workchain=workchain, + shard=shard, + seqno=seqno, + root_hash=root_hash, + file_hash=file_hash, + ) class BlockHeader(BaseModel): + type: Literal["blocks.header"] = Field(alias="@type") id: BlockId global_id: int version: int @@ -61,28 +70,28 @@ class BlockHeader(BaseModel): prev_blocks: List[BlockId] def build(tl_obj: dict): - check_tonlib_type(tl_obj, 'blocks.header') + check_tonlib_type(tl_obj, "blocks.header") print(tl_obj) return BlockHeader( - id=BlockId.build(tl_obj['id']), - global_id=tl_obj['global_id'], - version=tl_obj['version'], - flags=tl_obj.get('flags', 0), - after_merge=tl_obj['after_merge'], - after_split=tl_obj['after_split'], - before_split=tl_obj['before_split'], - want_merge=tl_obj['want_merge'], - want_split=tl_obj['want_split'], - validator_list_hash_short=tl_obj['validator_list_hash_short'], - catchain_seqno=tl_obj['catchain_seqno'], - min_ref_mc_seqno=tl_obj['min_ref_mc_seqno'], - is_key_block=tl_obj['is_key_block'], - prev_key_block_seqno=tl_obj['prev_key_block_seqno'], - start_lt=tl_obj['start_lt'], - end_lt=tl_obj['end_lt'], - gen_utime=tl_obj['gen_utime'], - vert_seqno=tl_obj.get('vert_seqno', 0), - prev_blocks=(BlockId.build(p) for p in tl_obj['prev_blocks']) + id=BlockId.build(tl_obj["id"]), + global_id=tl_obj["global_id"], + version=tl_obj["version"], + flags=tl_obj.get("flags", 0), + after_merge=tl_obj["after_merge"], + after_split=tl_obj["after_split"], + before_split=tl_obj["before_split"], + want_merge=tl_obj["want_merge"], + want_split=tl_obj["want_split"], + validator_list_hash_short=tl_obj["validator_list_hash_short"], + catchain_seqno=tl_obj["catchain_seqno"], + min_ref_mc_seqno=tl_obj["min_ref_mc_seqno"], + is_key_block=tl_obj["is_key_block"], + prev_key_block_seqno=tl_obj["prev_key_block_seqno"], + start_lt=tl_obj["start_lt"], + end_lt=tl_obj["end_lt"], + gen_utime=tl_obj["gen_utime"], + vert_seqno=tl_obj.get("vert_seqno", 0), + prev_blocks=(BlockId.build(p) for p in tl_obj["prev_blocks"]), ) @@ -93,34 +102,34 @@ class SmartContract(BaseModel): last_transaction_lt: Optional[int] last_transaction_hash: Optional[str] frozen_hash: Optional[str] - state: Literal['active', 'frozen', 'uninitialized'] + state: Literal["active", "frozen", "uninitialized"] contract_type: Optional[str] contract_extracted_data: Optional[dict] block_id: BlockId def build(tl_obj: dict): - check_tonlib_type(tl_obj, 'raw.fullAccountState') + check_tonlib_type(tl_obj, "raw.fullAccountState") - balance = int(tl_obj['balance']) if int(tl_obj['balance']) > 0 else 0 + balance = int(tl_obj["balance"]) if int(tl_obj["balance"]) > 0 else 0 state = address_state(tl_obj) - block_id = BlockId.build(tl_obj['block_id']) + block_id = BlockId.build(tl_obj["block_id"]) obj = SmartContract(balance=balance, state=state, block_id=block_id) - if len(tl_obj['code']): - obj.code = tl_obj['code'] - if len(tl_obj['data']): - obj.data = tl_obj['data'] - if int(tl_obj['last_transaction_id']['lt']): - obj.last_transaction_lt = int(tl_obj['last_transaction_id']['lt']) - obj.last_transaction_hash = tl_obj['last_transaction_id']['hash'] - if len(tl_obj['frozen_hash']): - obj.frozen_hash = tl_obj['frozen_hash'] - - ci = sha256(tl_obj['code']) + if len(tl_obj["code"]): + obj.code = tl_obj["code"] + if len(tl_obj["data"]): + obj.data = tl_obj["data"] + if int(tl_obj["last_transaction_id"]["lt"]): + obj.last_transaction_lt = int(tl_obj["last_transaction_id"]["lt"]) + obj.last_transaction_hash = tl_obj["last_transaction_id"]["hash"] + if len(tl_obj["frozen_hash"]): + obj.frozen_hash = tl_obj["frozen_hash"] + + ci = sha256(tl_obj["code"]) if ci in known_wallets: wallet_handler = known_wallets[ci] - obj.contract_type = wallet_handler['type'] + obj.contract_type = wallet_handler["type"] obj.contract_extracted_data = {} wallet_handler["data_extractor"](obj.contract_extracted_data, tl_obj) @@ -132,65 +141,129 @@ class AdressUserFriendly(BaseModel): b64url: str def build(raw: dict): - return AdressUserFriendly(b64=raw['b64'], b64url=raw['b64url']) + return AdressUserFriendly(b64=raw["b64"], b64url=raw["b64url"]) class AddressForms(BaseModel): raw_form: str - bounceable: AdressUserFriendly + bounceable: AdressUserFriendly non_bounceable: AdressUserFriendly given_type: Literal["friendly_bounceable", "friendly_non_bounceable", "raw_form"] test_only: bool def build(raw: dict): - return AddressForms(raw_form=raw['raw_form'], - bounceable=AdressUserFriendly.build(raw['bounceable']), - non_bounceable=AdressUserFriendly.build(raw['non_bounceable']), - given_type=raw['given_type'], - test_only=raw['test_only'] + return AddressForms( + raw_form=raw["raw_form"], + bounceable=AdressUserFriendly.build(raw["bounceable"]), + non_bounceable=AdressUserFriendly.build(raw["non_bounceable"]), + given_type=raw["given_type"], + test_only=raw["test_only"], ) class MasterchainInfo(BaseModel): + type: Literal["blocks.masterchainInfo"] = Field(alias="@type") last: BlockId state_root_hash: str init: BlockId def build(tl_obj: dict): - check_tonlib_type(tl_obj, 'blocks.masterchainInfo') + check_tonlib_type(tl_obj, "blocks.masterchainInfo") - return MasterchainInfo(last=BlockId.build(tl_obj['last']), - init=BlockId.build(tl_obj['init']), - state_root_hash=tl_obj['state_root_hash'] + return MasterchainInfo( + last=BlockId.build(tl_obj["last"]), + init=BlockId.build(tl_obj["init"]), + state_root_hash=tl_obj["state_root_hash"], ) +class BlockSignature(BaseModel): + type: Literal["blocks.signature"] = Field(alias="@type") + node_id_short: str + signature: str + + +class MasterchainSignatures(BaseModel): + type: Literal["blocks.blockSignatures"] = Field(alias="@type") + id: BlockId + signatures: List[BlockSignature] + + +class Proof(BaseModel): + type: Literal["blocks.blockLinkBack"] = Field(alias="@type") + to_key_block: bool + from_id: BlockId = Field(alias="from") + to: BlockId + dest_proof: str + proof: str + state_proof: str + + +class ShardBlockProof(BaseModel): + type: Literal["blocks.shardBlockProof"] = Field(alias="@type") + from_id: BlockId = Field(alias="from") + mc_id: BlockId + links: List[str] + mc_proof: List[Proof] + + +class Shards(BaseModel): + type: Literal["blocks.shards"] = Field(alias="@type") + shards: List[BlockId] + + +class ConsensusBlock(BaseModel): + consensus_block: int + timestamp: int + + +class ShortTransaction(BaseModel): + type: Literal["blocks.shortTxId"] = Field(alias="@type") + mode: int + account: str + lt: str + hash: str + + +class ShortTransactions(BaseModel): + type: Literal["blocks.transactions"] = Field(alias="@type") + id: BlockId + req_count: int + incomplete: bool + transactions: List[ShortTransaction] + + class ExternalMessage(BaseModel): msg_hash: str def build(tl_obj: dict): - check_tonlib_type(tl_obj, 'raw.extMessageInfo') + check_tonlib_type(tl_obj, "raw.extMessageInfo") - return ExternalMessage(msg_hash=tl_obj['hash']) + return ExternalMessage(msg_hash=tl_obj["hash"]) class SerializedBoc(BaseModel): boc: str def build_from_config(tl_obj: dict): - check_tonlib_type(tl_obj, 'configInfo') + check_tonlib_type(tl_obj, "configInfo") - return SerializedBoc(boc=tl_obj['config']['bytes']) + return SerializedBoc(boc=tl_obj["config"]["bytes"]) class MsgDataRaw(BaseModel): - body: str + body: str init_state: str def build(tl_obj: dict): - check_tonlib_type(tl_obj, 'msg.dataRaw') + check_tonlib_type(tl_obj, "msg.dataRaw") + + return MsgDataRaw(body=tl_obj["body"], init_state=tl_obj["init_state"]) - return MsgDataRaw(body=tl_obj['body'], init_state=tl_obj['init_state']) + +class MsgDataText(BaseModel): + type: Literal["msg.dataText"] = Field(alias="@type") + text: str class Message(BaseModel): @@ -201,23 +274,24 @@ class Message(BaseModel): ihr_fee: int created_lt: int body_hash: str - msg_data: MsgDataRaw + msg_data: Union[MsgDataRaw, MsgDataText] comment: Optional[str] op: Optional[int] def build(tl_obj: dict): - check_tonlib_type(tl_obj, 'raw.message') - - return Message(source=tl_obj['source'], - destination=tl_obj['destination'], - value=int(tl_obj['value']), - fwd_fee=int(tl_obj['fwd_fee']), - ihr_fee=int(tl_obj['ihr_fee']), - created_lt=int(tl_obj['created_lt']), - body_hash=tl_obj['body_hash'], - msg_data=MsgDataRaw.build(tl_obj['msg_data']), - comment=tl_obj.get('comment'), - op=tl_obj.get('op'), + check_tonlib_type(tl_obj, "raw.message") + + return Message( + source=tl_obj["source"], + destination=tl_obj["destination"], + value=int(tl_obj["value"]), + fwd_fee=int(tl_obj["fwd_fee"]), + ihr_fee=int(tl_obj["ihr_fee"]), + created_lt=int(tl_obj["created_lt"]), + body_hash=tl_obj["body_hash"], + msg_data=MsgDataRaw.build(tl_obj["msg_data"]), + comment=tl_obj.get("comment"), + op=tl_obj.get("op"), ) @@ -226,9 +300,18 @@ class TransactionId(BaseModel): hash: str def build(tl_obj: dict): - check_tonlib_type(tl_obj, 'internal.transactionId') + check_tonlib_type(tl_obj, "internal.transactionId") + + return TransactionId(lt=int(tl_obj["lt"]), hash=tl_obj["hash"]) + - return TransactionId(lt=int(tl_obj['lt']), hash=tl_obj['hash']) +class TransactionWAddressId(TransactionId): + account_address: str + + +class Address(BaseModel): + type: Literal["accountAddress"] = Field(alias="@type") + account_address: str class Transaction(BaseModel): @@ -243,15 +326,156 @@ class Transaction(BaseModel): out_msgs: List[Message] def build(tl_obj: dict): - check_tonlib_type(tl_obj, 'raw.transaction') - - return Transaction(utime=int(tl_obj['utime']), - data=tl_obj['data'], - hash=tl_obj['transaction_id']['hash'], - lt=tl_obj['transaction_id']['lt'], - fee=int(tl_obj['fee']), - storage_fee=int(tl_obj['storage_fee']), - other_fee=int(tl_obj['other_fee']), - in_msg=Message.build(tl_obj.get('in_msg')) if tl_obj.get('in_msg') else None, - out_msgs=[Message.build(m) for m in tl_obj['out_msgs']] + check_tonlib_type(tl_obj, "raw.transaction") + + return Transaction( + utime=int(tl_obj["utime"]), + data=tl_obj["data"], + hash=tl_obj["transaction_id"]["hash"], + lt=tl_obj["transaction_id"]["lt"], + fee=int(tl_obj["fee"]), + storage_fee=int(tl_obj["storage_fee"]), + other_fee=int(tl_obj["other_fee"]), + in_msg=( + Message.build(tl_obj.get("in_msg")) if tl_obj.get("in_msg") else None + ), + out_msgs=[Message.build(m) for m in tl_obj["out_msgs"]], ) + + +RawResultT = TypeVar("RawResultT") + + +class RawTransaction(GenericModel, Generic[RawResultT]): + type: Literal["raw.transaction"] = Field(alias="@type") + address: Address + utime: int + data: str + transaction_id: RawResultT + fee: int + storage_fee: int + other_fee: int + in_msg: Optional[Message] + out_msgs: List[Message] + + +class ShortTransaction(BaseModel): + type: Literal["blocks.shortTxId"] = Field(alias="@type") + mode: int + account: str + lt: str + hash: str + + +class ShortTransactions(BaseModel): + type: Literal["blocks.transactions"] = Field(alias="@type") + id: BlockId + req_count: int + incomplete: bool + transactions: List[ShortTransaction] + + +class TVMCell(BaseModel): + type: Literal["tvm.cell"] = Field(alias="@type") + bytes: str + + +class ConfigInfo(BaseModel): + type: Literal["configInfo"] = Field(alias="@type") + config: TVMCell + extra: str = Field(alias="@extra") + + +class TvmStackEntry(BaseModel): + type: str = Field(alias="@type") + + +class TvmTuple(BaseModel): + type: Literal["tvm.tuple"] = Field(alias="@type") + elements: List[TvmStackEntry] + + +class BlockIdExt(BaseModel): + type: Literal["smc.blockIdExt"] = Field(alias="@type") + workchain: int + shard: str + seqno: int + root_hash: str + file_hash: str + + +TVMStackEntryType = Literal["cell", "slice", "num", "tuple", "list"] + + +class AccountStateWallet(BaseModel): + type: str = Field("wallet.version.accountState", alias="@type") + wallet_id: str + seqno: int + + +class AccountStateRow(BaseModel): + type: str = Field("raw.accountState", alias="@type") + code: str + data: str + frozen_hash: str + + +class AccountStateUninited(BaseModel): + type: str = Field("uninited.accountState", alias="@type") + frozen_hash: str + + +class AddressShort(BaseModel): + type: str = Field(alias="@type") + account_address: str + + +class JettonMasterData(BaseModel): + total_supply: int + mintable: bool + admin_address: str + + class JettonContent(BaseModel): + type: str = Field(alias="@type") + + class Data(BaseModel): + image: str + name: str + symbol: str + description: str + decimals: str + + data: Data + + jetton_content: JettonContent + jetton_wallet_code: str + contract_type: str = Field("jetton_master") + + +class JettonWalletData(BaseModel): + balance: int + owner: str + jetton: str + jetton_wallet_code: str + contract_type: str = Field("jetton_wallet") + + +class NftContent(BaseModel): + type: str = Field(alias="@type") + data: str + + +class NftCollectionData(BaseModel): + next_item_index: int + collection_content: NftContent + owner_address: str + contract_type: str = Field("nft_collection") + + +class NftItemData(BaseModel): + init: bool + index: int + owner_address: str + collection_address: str + content: NftContent + contract_type: str = Field("nft_item")