Merge pull request #286 from InjectiveLabs/feat/initialize_tokens_from_chain_denoms

Feat/initialize tokens from chain denoms

Author: aarmoa

examples/chain_client/50_SpendableBalances.py

@@ -0,0 +1,16 @@
+import asyncio
+
+from pyinjective.async_client import AsyncClient
+from pyinjective.core.network import Network
+
+
+async def main() -> None:
+    network = Network.testnet()
+    client = AsyncClient(network)
+    address = "inj1cml96vmptgw99syqrrz8az79xer2pcgp0a885r"
+    spendable_balances = await client.fetch_spendable_balances(address=address)
+    print(spendable_balances)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

examples/chain_client/51_SpendableBalancesByDenom.py

@@ -0,0 +1,17 @@
+import asyncio
+
+from pyinjective.async_client import AsyncClient
+from pyinjective.core.network import Network
+
+
+async def main() -> None:
+    network = Network.testnet()
+    client = AsyncClient(network)
+    address = "inj1cml96vmptgw99syqrrz8az79xer2pcgp0a885r"
+    denom = "inj"
+    spendable_balances = await client.fetch_spendable_balances_by_denom(address=address, denom=denom)
+    print(spendable_balances)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

examples/chain_client/52_TotalSupply.py

@@ -0,0 +1,18 @@
+import asyncio
+
+from pyinjective.async_client import AsyncClient
+from pyinjective.client.model.pagination import PaginationOption
+from pyinjective.core.network import Network
+
+
+async def main() -> None:
+    network = Network.testnet()
+    client = AsyncClient(network)
+    total_supply = await client.fetch_total_supply(
+        pagination=PaginationOption(limit=10),
+    )
+    print(total_supply)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

examples/chain_client/53_SupplyOf.py

@@ -0,0 +1,15 @@
+import asyncio
+
+from pyinjective.async_client import AsyncClient
+from pyinjective.core.network import Network
+
+
+async def main() -> None:
+    network = Network.testnet()
+    client = AsyncClient(network)
+    supply_of = await client.fetch_supply_of(denom="inj")
+    print(supply_of)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

examples/chain_client/54_DenomMetadata.py

@@ -0,0 +1,16 @@
+import asyncio
+
+from pyinjective.async_client import AsyncClient
+from pyinjective.core.network import Network
+
+
+async def main() -> None:
+    network = Network.testnet()
+    client = AsyncClient(network)
+    denom = "factory/inj107aqkjc3t5r3l9j4n9lgrma5tm3jav8qgppz6m/position"
+    metadata = await client.fetch_denom_metadata(denom=denom)
+    print(metadata)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

examples/chain_client/55_DenomsMetadata.py

@@ -0,0 +1,18 @@
+import asyncio
+
+from pyinjective.async_client import AsyncClient
+from pyinjective.client.model.pagination import PaginationOption
+from pyinjective.core.network import Network
+
+
+async def main() -> None:
+    network = Network.testnet()
+    client = AsyncClient(network)
+    denoms = await client.fetch_denoms_metadata(
+        pagination=PaginationOption(limit=10),
+    )
+    print(denoms)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

examples/chain_client/56_DenomOwners.py

@@ -0,0 +1,20 @@
+import asyncio
+
+from pyinjective.async_client import AsyncClient
+from pyinjective.client.model.pagination import PaginationOption
+from pyinjective.core.network import Network
+
+
+async def main() -> None:
+    network = Network.testnet()
+    client = AsyncClient(network)
+    denom = "inj"
+    owners = await client.fetch_denom_owners(
+        denom=denom,
+        pagination=PaginationOption(limit=10),
+    )
+    print(owners)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

examples/chain_client/57_SendEnabled.py

@@ -0,0 +1,20 @@
+import asyncio
+
+from pyinjective.async_client import AsyncClient
+from pyinjective.client.model.pagination import PaginationOption
+from pyinjective.core.network import Network
+
+
+async def main() -> None:
+    network = Network.testnet()
+    client = AsyncClient(network)
+    denom = "inj"
+    enabled = await client.fetch_send_enabled(
+        denoms=[denom],
+        pagination=PaginationOption(limit=10),
+    )
+    print(enabled)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

pyinjective/async_client.py

@@ -1,4 +1,5 @@
 import asyncio
+import base64
 import time
 from copy import deepcopy
 from decimal import Decimal
@@ -522,6 +523,42 @@ async def get_bank_balance(self, address: str, denom: str):
     async def fetch_bank_balance(self, address: str, denom: str) -> Dict[str, Any]:
         return await self.bank_api.fetch_balance(account_address=address, denom=denom)
 
+    async def fetch_spendable_balances(
+        self,
+        address: str,
+        pagination: Optional[PaginationOption] = None,
+    ) -> Dict[str, Any]:
+        return await self.bank_api.fetch_spendable_balances(account_address=address, pagination=pagination)
+
+    async def fetch_spendable_balances_by_denom(
+        self,
+        address: str,
+        denom: str,
+    ) -> Dict[str, Any]:
+        return await self.bank_api.fetch_spendable_balances_by_denom(account_address=address, denom=denom)
+
+    async def fetch_total_supply(self, pagination: Optional[PaginationOption] = None) -> Dict[str, Any]:
+        return await self.bank_api.fetch_total_supply(pagination=pagination)
+
+    async def fetch_supply_of(self, denom: str) -> Dict[str, Any]:
+        return await self.bank_api.fetch_supply_of(denom=denom)
+
+    async def fetch_denom_metadata(self, denom: str) -> Dict[str, Any]:
+        return await self.bank_api.fetch_denom_metadata(denom=denom)
+
+    async def fetch_denoms_metadata(self, pagination: Optional[PaginationOption] = None) -> Dict[str, Any]:
+        return await self.bank_api.fetch_denoms_metadata(pagination=pagination)
+
+    async def fetch_denom_owners(self, denom: str, pagination: Optional[PaginationOption] = None) -> Dict[str, Any]:
+        return await self.bank_api.fetch_denom_owners(denom=denom, pagination=pagination)
+
+    async def fetch_send_enabled(
+        self,
+        denoms: Optional[List[str]] = None,
+        pagination: Optional[PaginationOption] = None,
+    ) -> Dict[str, Any]:
+        return await self.bank_api.fetch_send_enabled(denoms=denoms, pagination=pagination)
+
     # Injective Exchange client methods
 
     # Auction RPC
@@ -2461,6 +2498,51 @@ async def composer(self):
             tokens=await self.all_tokens(),
         )
 
+    async def initialize_tokens_from_chain_denoms(self):
+        # force initialization of markets and tokens
+        await self.all_tokens()
+
+        all_denoms_metadata = []
+
+        query_result = await self.fetch_denoms_metadata()
+
+        all_denoms_metadata.extend(query_result.get("metadatas", []))
+        next_key = query_result.get("pagination", {}).get("nextKey", "")
+
+        while next_key != "":
+            query_result = await self.fetch_denoms_metadata(pagination=PaginationOption(key=next_key))
+
+            all_denoms_metadata.extend(query_result.get("metadatas", []))
+            result_next_key = query_result.get("pagination", {}).get("nextKey", "")
+            next_key = base64.b64decode(result_next_key).decode()
+
+        for token_metadata in all_denoms_metadata:
+            symbol = token_metadata["symbol"]
+            denom = token_metadata["base"]
+
+            if denom != "" and symbol != "" and denom not in self._tokens_by_denom:
+                name = token_metadata["name"] or symbol
+                decimals = max({denom_unit["exponent"] for denom_unit in token_metadata["denomUnits"]})
+
+                unique_symbol = denom
+                for symbol_candidate in [symbol, name]:
+                    if symbol_candidate not in self._tokens_by_symbol:
+                        unique_symbol = symbol_candidate
+                        break
+
+                token = Token(
+                    name=name,
+                    symbol=symbol,
+                    denom=denom,
+                    address="",
+                    decimals=decimals,
+                    logo=token_metadata["uri"],
+                    updated=-1,
+                )
+
+                self._tokens_by_denom[denom] = token
+                self._tokens_by_symbol[unique_symbol] = token
+
     async def _initialize_tokens_and_markets(self):
         spot_markets = dict()
         derivative_markets = dict()

pyinjective/client/chain/grpc/chain_grpc_bank_api.py

@@ -1,7 +1,8 @@
-from typing import Any, Callable, Dict
+from typing import Any, Callable, Dict, List, Optional
 
 from grpc.aio import Channel
 
+from pyinjective.client.model.pagination import PaginationOption
 from pyinjective.proto.cosmos.bank.v1beta1 import query_pb2 as bank_query_pb, query_pb2_grpc as bank_query_grpc
 from pyinjective.utils.grpc_api_request_assistant import GrpcApiRequestAssistant
 
@@ -29,11 +30,86 @@ async def fetch_balances(self, account_address: str) -> Dict[str, Any]:
 
         return response
 
-    async def fetch_total_supply(self) -> Dict[str, Any]:
-        request = bank_query_pb.QueryTotalSupplyRequest()
+    async def fetch_spendable_balances(
+        self,
+        account_address: str,
+        pagination: Optional[PaginationOption] = None,
+    ) -> Dict[str, Any]:
+        pagination_request = None
+        if pagination is not None:
+            pagination_request = pagination.create_pagination_request()
+        request = bank_query_pb.QuerySpendableBalancesRequest(
+            address=account_address,
+            pagination=pagination_request,
+        )
+        response = await self._execute_call(call=self._stub.SpendableBalances, request=request)
+
+        return response
+
+    async def fetch_spendable_balances_by_denom(
+        self,
+        account_address: str,
+        denom: str,
+    ) -> Dict[str, Any]:
+        request = bank_query_pb.QuerySpendableBalanceByDenomRequest(
+            address=account_address,
+            denom=denom,
+        )
+        response = await self._execute_call(call=self._stub.SpendableBalanceByDenom, request=request)
+
+        return response
+
+    async def fetch_total_supply(self, pagination: Optional[PaginationOption] = None) -> Dict[str, Any]:
+        pagination_request = None
+        if pagination is not None:
+            pagination_request = pagination.create_pagination_request()
+        request = bank_query_pb.QueryTotalSupplyRequest(pagination=pagination_request)
         response = await self._execute_call(call=self._stub.TotalSupply, request=request)
 
         return response
 
+    async def fetch_supply_of(self, denom: str) -> Dict[str, Any]:
+        request = bank_query_pb.QuerySupplyOfRequest(denom=denom)
+        response = await self._execute_call(call=self._stub.SupplyOf, request=request)
+
+        return response
+
+    async def fetch_denom_metadata(self, denom: str) -> Dict[str, Any]:
+        request = bank_query_pb.QueryDenomMetadataRequest(denom=denom)
+        response = await self._execute_call(call=self._stub.DenomMetadata, request=request)
+
+        return response
+
+    async def fetch_denoms_metadata(self, pagination: Optional[PaginationOption] = None) -> Dict[str, Any]:
+        pagination_request = None
+        if pagination is not None:
+            pagination_request = pagination.create_pagination_request()
+        request = bank_query_pb.QueryDenomsMetadataRequest(pagination=pagination_request)
+        response = await self._execute_call(call=self._stub.DenomsMetadata, request=request)
+
+        return response
+
+    async def fetch_denom_owners(self, denom: str, pagination: Optional[PaginationOption] = None) -> Dict[str, Any]:
+        pagination_request = None
+        if pagination is not None:
+            pagination_request = pagination.create_pagination_request()
+        request = bank_query_pb.QueryDenomOwnersRequest(denom=denom, pagination=pagination_request)
+        response = await self._execute_call(call=self._stub.DenomOwners, request=request)
+
+        return response
+
+    async def fetch_send_enabled(
+        self,
+        denoms: Optional[List[str]] = None,
+        pagination: Optional[PaginationOption] = None,
+    ) -> Dict[str, Any]:
+        pagination_request = None
+        if pagination is not None:
+            pagination_request = pagination.create_pagination_request()
+        request = bank_query_pb.QuerySendEnabledRequest(denoms=denoms, pagination=pagination_request)
+        response = await self._execute_call(call=self._stub.SendEnabled, request=request)
+
+        return response
+
     async def _execute_call(self, call: Callable, request) -> Dict[str, Any]:
         return await self._assistant.execute_call(call=call, request=request)