From 2a1a5c8a9d667068fc350149d21be069fb0d358d Mon Sep 17 00:00:00 2001 From: fishronsage Date: Mon, 6 May 2024 23:17:46 +0800 Subject: [PATCH] basic implementation of fungible asset client --- aptos_sdk/aptos_token_client.py | 261 +++++++++++++++++++++++++++++++- examples/your_fungible_asset.py | 247 ++++++++++++++++++++++++++++++ 2 files changed, 507 insertions(+), 1 deletion(-) create mode 100644 examples/your_fungible_asset.py diff --git a/aptos_sdk/aptos_token_client.py b/aptos_sdk/aptos_token_client.py index 96e8c52..15ae83d 100644 --- a/aptos_sdk/aptos_token_client.py +++ b/aptos_sdk/aptos_token_client.py @@ -3,7 +3,7 @@ from __future__ import annotations -from typing import Any, List, Tuple +from typing import Any, List, Optional, Tuple from .account import Account from .account_address import AccountAddress @@ -298,6 +298,44 @@ def parse(resource: dict[str, Any]) -> PropertyMap: return PropertyMap(properties) +class FAConcurrentSupply: + value: int + max_value: int + + struct_tag = "0x1::fungible_asset::ConcurrentSupply" + + def __init__(self, value: int, max_value: int) -> None: + self.value = value + self.max_value = max_value + + def __str__(self) -> str: + return f"FAConcurrentSupply[value: {self.value}, max_value: {self.max_value}]" + + @staticmethod + def parse(resource: dict[str, Any]) -> FAConcurrentSupply: + return FAConcurrentSupply( + int(resource["current"]["value"]), int(resource["current"]["max_value"]) + ) + + +class FungibleStore: + balance: int + frozen: bool + + struct_tag = "0x1::fungible_asset::FungibleStore" + + def __init__(self, balance: int, frozen: bool) -> None: + self.balance = balance + self.frozen = frozen + + def __str__(self): + return f"FungibleStore[balance: {self.balance}, frozen: {self.frozen}]" + + @staticmethod + def parse(resource: dict[str, Any]) -> FungibleStore: + return FungibleStore(int(resource["balance"]), resource["frozen"]) + + class ReadObject: resource_map: dict[str, Any] = { Collection.struct_tag: Collection, @@ -305,6 +343,8 @@ class ReadObject: PropertyMap.struct_tag: PropertyMap, Royalty.struct_tag: Royalty, Token.struct_tag: Token, + FAConcurrentSupply.struct_tag: FAConcurrentSupply, + FungibleStore.struct_tag: FungibleStore, } resources: dict[Any, Any] @@ -631,3 +671,222 @@ async def tokens_minted_from_transaction( continue mints.append(AccountAddress.from_str_relaxed(event["data"]["token"])) return mints + + +class FungibleAssetClient: + """A wrapper around reading and mutating Fungible Assets""" + + def __init__(self, rest_client: RestClient): + self.client = rest_client + + async def __primary_store_view( + self, + function: str, + args: List[TransactionArgument], + ledger_version: Optional[int] = None, + ) -> Any: + module = "0x1::primary_fungible_store" + ty_args = [TypeTag(StructTag.from_str("0x1::fungible_asset::Metadata"))] + return await self.client.view_bcs_payload( + module, function, ty_args, args, ledger_version + ) + + async def __metadata_view( + self, + function: str, + args: List[TransactionArgument], + ledger_version: Optional[int] = None, + ) -> Any: + module = "0x1::fungible_asset" + ty_args = [TypeTag(StructTag.from_str("0x1::fungible_asset::Metadata"))] + return await self.client.view_bcs_payload( + module, function, ty_args, args, ledger_version + ) + + async def read_object(self, address: AccountAddress) -> ReadObject: + resources = {} + read_resources = await self.client.account_resources(address) + for resource in read_resources: + if resource["type"] in ReadObject.resource_map: + resource_obj = ReadObject.resource_map[resource["type"]] + resources[resource_obj] = resource_obj.parse(resource["data"]) + return ReadObject(resources) + + async def supply( + self, metadata_address: AccountAddress, ledger_version: Optional[int] = None + ) -> int: + """Get the current supply from the metadata object.""" + resp = await self.__metadata_view( + "supply", + [ + TransactionArgument(metadata_address, Serializer.struct), + ], + ledger_version, + ) + return int(resp[0]["vec"][0]) + + async def maximum( + self, metadata_address: AccountAddress, ledger_version: Optional[int] = None + ) -> int: + """Get the maximum supply from the metadata object. If supply is unlimited (or set explicitly to MAX_U128), none is returned.""" + resp = await self.__metadata_view( + "maximum", + [ + TransactionArgument(metadata_address, Serializer.struct), + ], + ledger_version, + ) + return int(resp[0]["vec"][0]) + + async def name( + self, metadata_address: AccountAddress, ledger_version: Optional[int] = None + ) -> str: + """Get the name of the fungible asset from the metadata object.""" + resp = await self.__metadata_view( + "name", + [ + TransactionArgument(metadata_address, Serializer.struct), + ], + ledger_version, + ) + return resp[0] + + async def symbol( + self, metadata_address: AccountAddress, ledger_version: Optional[int] = None + ) -> str: + """Get the symbol of the fungible asset from the metadata object.""" + resp = await self.__metadata_view( + "symbol", + [ + TransactionArgument(metadata_address, Serializer.struct), + ], + ledger_version, + ) + return resp[0] + + async def decimals( + self, metadata_address: AccountAddress, ledger_version: Optional[int] = None + ) -> int: + """Get the decimals from the metadata object.""" + resp = await self.__metadata_view( + "decimals", + [ + TransactionArgument(metadata_address, Serializer.struct), + ], + ledger_version, + ) + return int(resp[0]) + + async def icon_uri( + self, metadata_address: AccountAddress, ledger_version: Optional[int] = None + ) -> str: + """Get the icon uri from the metadata object.""" + resp = await self.__metadata_view( + "icon_uri", + [ + TransactionArgument(metadata_address, Serializer.struct), + ], + ledger_version, + ) + return resp[0] + + async def project_uri( + self, metadata_address: AccountAddress, ledger_version: Optional[int] = None + ) -> str: + """Get the project uri from the metadata object.""" + resp = await self.__metadata_view( + "project_uri", + [ + TransactionArgument(metadata_address, Serializer.struct), + ], + ledger_version, + ) + return resp[0] + + async def store_metadata( + self, address: AccountAddress, ledger_version: Optional[int] = None + ) -> str: + """Return the underlying metadata object.""" + resp = await self.client.view_bcs_payload( + "0x1::fungible_asset", + "store_metadata", + [TypeTag(StructTag.from_str("0x1::fungible_asset::FungibleStore"))], + [TransactionArgument(address, Serializer.struct)], + ledger_version, + ) + return resp[0] + + async def transfer( + self, + sender: Account, + metadata_address: AccountAddress, + receiver_address: AccountAddress, + amount: int, + sequence_number: Optional[int] = None, + ) -> str: + """Transfer amount of fungible asset from sender's primary store to receiver's primary store.""" + payload = EntryFunction.natural( + "0x1::primary_fungible_store", + "transfer", + [TypeTag(StructTag.from_str("0x1::fungible_asset::Metadata"))], + [ + TransactionArgument(metadata_address, Serializer.struct), + TransactionArgument(receiver_address, Serializer.struct), + TransactionArgument(amount, Serializer.u64), + ], + ) + signed_transaction = await self.client.create_bcs_signed_transaction( + sender, TransactionPayload(payload), sequence_number=sequence_number + ) + return await self.client.submit_bcs_transaction(signed_transaction) + + async def balance( + self, + metadata_address: AccountAddress, + address: AccountAddress, + ledger_version: Optional[int] = None, + ) -> int: + """Get the balance of account's primary store.""" + resp = await self.__primary_store_view( + "balance", + [ + TransactionArgument(address, Serializer.struct), + TransactionArgument(metadata_address, Serializer.struct), + ], + ledger_version, + ) + return int(resp[0]) + + async def is_frozen( + self, + metadata_address: AccountAddress, + address: AccountAddress, + ledger_version: Optional[int] = None, + ) -> bool: + """Return whether the given account's primary store is frozen.""" + resp = await self.__primary_store_view( + "is_frozen", + [ + TransactionArgument(address, Serializer.struct), + TransactionArgument(metadata_address, Serializer.struct), + ], + ledger_version, + ) + return resp[0] + + async def primary_store_address( + self, + metadata_address, + address: AccountAddress, + ledger_version: Optional[int] = None, + ) -> str: + """Get the address of the primary store for the given account.""" + resp = await self.__primary_store_view( + "primary_store_address", + [ + TransactionArgument(address, Serializer.struct), + TransactionArgument(metadata_address, Serializer.struct), + ], + ledger_version, + ) + return resp[0] diff --git a/examples/your_fungible_asset.py b/examples/your_fungible_asset.py new file mode 100644 index 0000000..e0cbd11 --- /dev/null +++ b/examples/your_fungible_asset.py @@ -0,0 +1,247 @@ +# Copyright © Aptos Foundation +# SPDX-License-Identifier: Apache-2.0 + +import asyncio +import os +import sys + +from aptos_sdk.account import Account, AccountAddress +from aptos_sdk.aptos_cli_wrapper import AptosCLIWrapper +from aptos_sdk.aptos_token_client import FungibleAssetClient +from aptos_sdk.async_client import FaucetClient, RestClient +from aptos_sdk.bcs import Serializer +from aptos_sdk.package_publisher import PackagePublisher +from aptos_sdk.transactions import ( + EntryFunction, + TransactionArgument, + TransactionPayload, +) + +from .common import FAUCET_URL, NODE_URL + + +# Admin forcefully transfers the newly created coin to the specified receiver address +async def transfer_coin( + rest_client: RestClient, + admin: Account, + from_addr: AccountAddress, + to_addr: AccountAddress, + amount: int, +) -> None: + payload = EntryFunction.natural( + f"{admin.address()}::fa_coin", + "transfer", + [], + [ + TransactionArgument(from_addr, Serializer.struct), + TransactionArgument(to_addr, Serializer.struct), + TransactionArgument(amount, Serializer.u64), + ], + ) + signed_txn = await rest_client.create_bcs_signed_transaction( + admin, TransactionPayload(payload) + ) + txn_hash = await rest_client.submit_bcs_transaction(signed_txn) + await rest_client.wait_for_transaction(txn_hash) + + +# Admin mint the newly created coin to the specified receiver address +async def mint_coin( + rest_client: RestClient, admin: Account, receiver: AccountAddress, amount: int +) -> None: + payload = EntryFunction.natural( + f"{admin.address()}::fa_coin", + "mint", + [], + [ + TransactionArgument(receiver, Serializer.struct), + TransactionArgument(amount, Serializer.u64), + ], + ) + signed_txn = await rest_client.create_bcs_signed_transaction( + admin, TransactionPayload(payload) + ) + txn_hash = await rest_client.submit_bcs_transaction(signed_txn) + await rest_client.wait_for_transaction(txn_hash) + + +# Admin burns the newly created coin from the specified receiver address +async def burn_coin( + rest_client: RestClient, admin: Account, receiver: AccountAddress, amount: int +) -> None: + payload = EntryFunction.natural( + f"{admin.address()}::fa_coin", + "burn", + [], + [ + TransactionArgument(receiver, Serializer.struct), + TransactionArgument(amount, Serializer.u64), + ], + ) + signed_txn = await rest_client.create_bcs_signed_transaction( + admin, TransactionPayload(payload) + ) + txn_hash = await rest_client.submit_bcs_transaction(signed_txn) + await rest_client.wait_for_transaction(txn_hash) + + +# Admin freezes the primary fungible store of the specified account +async def freeze( + rest_client: RestClient, admin: Account, target_addr: AccountAddress +) -> None: + payload = EntryFunction.natural( + f"{admin.address()}::fa_coin", + "freeze_account", + [], + [ + TransactionArgument(target_addr, Serializer.struct), + ], + ) + signed_txn = await rest_client.create_bcs_signed_transaction( + admin, TransactionPayload(payload) + ) + txn_hash = await rest_client.submit_bcs_transaction(signed_txn) + await rest_client.wait_for_transaction(txn_hash) + + +# Admin unfreezes the primary fungible store of the specified account +async def unfreeze( + rest_client: RestClient, admin: Account, target_addr: AccountAddress +) -> None: + payload = EntryFunction.natural( + f"{admin.address()}::fa_coin", + "unfreeze_account", + [], + [ + TransactionArgument(target_addr, Serializer.struct), + ], + ) + signed_txn = await rest_client.create_bcs_signed_transaction( + admin, TransactionPayload(payload) + ) + txn_hash = await rest_client.submit_bcs_transaction(signed_txn) + await rest_client.wait_for_transaction(txn_hash) + + +async def main(facoin_path: str): + alice = Account.generate() + bob = Account.generate() + charlie = Account.generate() + + print("=== Addresses ===") + print(f"Alice: {alice.address()}") + print(f"Bob: {bob.address()}") + print(f"Charlie: {charlie.address()}") + + rest_client = RestClient(NODE_URL) + faucet_client = FaucetClient(FAUCET_URL, rest_client) + + alice_fund = faucet_client.fund_account(alice.address(), 100_000_000) + bob_fund = faucet_client.fund_account(bob.address(), 100_000_000) + await asyncio.gather(*[alice_fund, bob_fund]) + + if AptosCLIWrapper.does_cli_exist(): + print("\n=== Compiling FACoin package locally ===") + AptosCLIWrapper.compile_package(facoin_path, {"FACoin": alice.address()}) + else: + input("\nUpdate the module with Alice's address, compile, and press enter.") + + # :!:>publish + module_path = os.path.join( + facoin_path, "build", "Examples", "bytecode_modules", "fa_coin.mv" + ) + with open(module_path, "rb") as f: + module = f.read() + + metadata_path = os.path.join( + facoin_path, "build", "Examples", "package-metadata.bcs" + ) + with open(metadata_path, "rb") as f: + metadata = f.read() + + print("\n===Publishing FACoin package===") + package_publisher = PackagePublisher(rest_client) + txn_hash = await package_publisher.publish_package(alice, metadata, [module]) + await rest_client.wait_for_transaction(txn_hash) + print("Transaction hash:", txn_hash) + # <:!:publish + + get_metadata_resp = await rest_client.view_bcs_payload( + f"{alice.address()}::fa_coin", "get_metadata", [], [] + ) + facoin_address = AccountAddress.from_str(get_metadata_resp[0]["inner"]) + print("FACoin address:", facoin_address) + + fa_client = FungibleAssetClient(rest_client) + + print( + "All the balances in this example refer to balance in primary fungible stores of each account." + ) + print( + f"Alice's initial FACoin balance: {await fa_client.balance(facoin_address, alice.address())}." + ) + print( + f"Bob's initial FACoin balance: {await fa_client.balance(facoin_address, bob.address())}." + ) + print( + f"Charlie's initial FACoin balance: {await fa_client.balance(facoin_address, charlie.address())}." + ) + + print("Alice mints Charlie 100 coins.") + await mint_coin(rest_client, alice, charlie.address(), 100) + + charlie_primary_store_addr = await fa_client.primary_store_address( + facoin_address, charlie.address() + ) + print(f"Charlie primary store address: {charlie_primary_store_addr}") + print( + f"Charlie's FACoin: {await fa_client.read_object(AccountAddress.from_str_relaxed(charlie_primary_store_addr))}" + ) + + print("Alice freeze Bob's account.") + await freeze(rest_client, alice, bob.address()) + + print( + "Alice as the admin forcefully transfers the newly minted coins of Charlie to Bob ignoring that Bob's account is frozen." + ) + await transfer_coin(rest_client, alice, charlie.address(), bob.address(), 100) + print( + f"Bob's updated FACoin balance: {await fa_client.balance(facoin_address, bob.address())}." + ) + print("Bob is frozen:", await fa_client.is_frozen(facoin_address, bob.address())) + + print("Alice unfreezes Bob's account.") + await unfreeze(rest_client, alice, bob.address()) + + print("Alice burns 50 coins from Bob.") + await burn_coin(rest_client, alice, bob.address(), 50) + print( + f"Bob's updated FACoin balance: {await fa_client.balance(facoin_address, bob.address())}." + ) + + print("Bob transfers 10 coins to Alice as the owner.") + await fa_client.transfer(bob, facoin_address, alice.address(), 10) + print( + f"Alice's updated FACoin balance: {await fa_client.balance(facoin_address, alice.address())}." + ) + print( + f"Bob's updated FACoin balance: {await fa_client.balance(facoin_address, bob.address())}." + ) + + print(f"Current FACoin's metadata: f{await fa_client.read_object(facoin_address)}") + print(f"Name: {await fa_client.name(facoin_address)}") + print(f"Supply: {await fa_client.supply(facoin_address)}") + print(f"Maximum: {await fa_client.maximum(facoin_address)}") + print(f"Decimals: {await fa_client.decimals(facoin_address)}") + print(f"Icon uri: {await fa_client.icon_uri(facoin_address)}") + print(f"Project uri: {await fa_client.project_uri(facoin_address)}") + + print("done.") + + +if __name__ == "__main__": + assert ( + len(sys.argv) == 2 + ), "Expecting an argument that points to the fa_coin directory." + + asyncio.run(main(sys.argv[1]))