(feat) Added all endpoint for chain bank module

Author: abel

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

@@ -522,6 +522,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

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)