(feat) Implemented all TokenFactory query module endpoints

Author: abel

examples/chain_client/68_AuthorityDenomMetadata.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)
+    metadata = await client.fetch_denom_authority_metadata(creator="inj1zvy8xrlhe7ex9scer868clfstdv7j6vz790kwa")
+    print(metadata)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

examples/chain_client/69_DenomsFromCreator.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)
+    denoms = await client.fetch_denoms_from_creator(creator="inj1maeyvxfamtn8lfyxpjca8kuvauuf2qeu6gtxm3")
+    print(denoms)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

examples/chain_client/70_TokenfactoryModuleState.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)
+    state = await client.fetch_tokenfactory_module_state()
+    print(state)
+
+
+if __name__ == "__main__":
+    asyncio.get_event_loop().run_until_complete(main())

pyinjective/async_client.py

@@ -13,6 +13,7 @@
 from pyinjective.client.chain.grpc.chain_grpc_auth_api import ChainGrpcAuthApi
 from pyinjective.client.chain.grpc.chain_grpc_authz_api import ChainGrpcAuthZApi
 from pyinjective.client.chain.grpc.chain_grpc_bank_api import ChainGrpcBankApi
+from pyinjective.client.chain.grpc.chain_grpc_token_factory_api import ChainGrpcTokenFactoryApi
 from pyinjective.client.chain.grpc.chain_grpc_wasm_api import ChainGrpcWasmApi
 from pyinjective.client.chain.grpc_stream.chain_grpc_chain_stream import ChainGrpcChainStream
 from pyinjective.client.indexer.grpc.indexer_grpc_account_api import IndexerGrpcAccountApi
@@ -178,6 +179,12 @@ def __init__(
                 metadata_query_provider=self._chain_cookie_metadata_requestor
             ),
         )
+        self.token_factory_api = ChainGrpcTokenFactoryApi(
+            channel=self.chain_channel,
+            metadata_provider=lambda: self.network.chain_metadata(
+                metadata_query_provider=self._chain_cookie_metadata_requestor
+            ),
+        )
         self.tx_api = TxGrpcApi(
             channel=self.chain_channel,
             metadata_provider=lambda: self.network.chain_metadata(
@@ -738,6 +745,24 @@ async def fetch_contracts_by_creator(
             pagination=pagination,
         )
 
+    # Token Factory module
+
+    async def fetch_denom_authority_metadata(
+        self,
+        creator: str,
+        sub_denom: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        return await self.token_factory_api.fetch_denom_authority_metadata(creator=creator, sub_denom=sub_denom)
+
+    async def fetch_denoms_from_creator(
+        self,
+        creator: str,
+    ) -> Dict[str, Any]:
+        return await self.token_factory_api.fetch_denoms_from_creator(creator=creator)
+
+    async def fetch_tokenfactory_module_state(self) -> Dict[str, Any]:
+        return await self.token_factory_api.fetch_tokenfactory_module_state()
+
     # Explorer RPC
 
     async def get_tx_by_hash(self, tx_hash: str):

pyinjective/client/chain/grpc/chain_grpc_token_factory_api.py

@@ -0,0 +1,51 @@
+from typing import Any, Callable, Dict, Optional
+
+from grpc.aio import Channel
+
+from pyinjective.proto.injective.tokenfactory.v1beta1 import (
+    query_pb2 as token_factory_query_pb,
+    query_pb2_grpc as token_factory_query_grpc,
+    tx_pb2_grpc as token_factory_tx_grpc,
+)
+from pyinjective.utils.grpc_api_request_assistant import GrpcApiRequestAssistant
+
+
+class ChainGrpcTokenFactoryApi:
+    def __init__(self, channel: Channel, metadata_provider: Callable):
+        self._query_stub = token_factory_query_grpc.QueryStub(channel)
+        self._tx_stub = token_factory_tx_grpc.MsgStub(channel)
+        self._assistant = GrpcApiRequestAssistant(metadata_provider=metadata_provider)
+
+    async def fetch_module_params(self) -> Dict[str, Any]:
+        request = token_factory_query_pb.QueryParamsRequest()
+        response = await self._execute_call(call=self._query_stub.Params, request=request)
+
+        return response
+
+    async def fetch_denom_authority_metadata(
+        self,
+        creator: str,
+        sub_denom: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        request = token_factory_query_pb.QueryDenomAuthorityMetadataRequest(
+            creator=creator,
+            sub_denom=sub_denom,
+        )
+        response = await self._execute_call(call=self._query_stub.DenomAuthorityMetadata, request=request)
+
+        return response
+
+    async def fetch_denoms_from_creator(self, creator: str) -> Dict[str, Any]:
+        request = token_factory_query_pb.QueryDenomsFromCreatorRequest(creator=creator)
+        response = await self._execute_call(call=self._query_stub.DenomsFromCreator, request=request)
+
+        return response
+
+    async def fetch_tokenfactory_module_state(self) -> Dict[str, Any]:
+        request = token_factory_query_pb.QueryModuleStateRequest()
+        response = await self._execute_call(call=self._query_stub.TokenfactoryModuleState, 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)

tests/client/chain/grpc/configurable_token_factory_query_servicer.py

@@ -0,0 +1,33 @@
+from collections import deque
+
+from pyinjective.proto.injective.tokenfactory.v1beta1 import (
+    query_pb2 as token_factory_query_pb,
+    query_pb2_grpc as token_factory_query_grpc,
+)
+
+
+class ConfigurableTokenFactoryQueryServicer(token_factory_query_grpc.QueryServicer):
+    def __init__(self):
+        super().__init__()
+        self.params_responses = deque()
+        self.denom_authority_metadata_responses = deque()
+        self.denoms_from_creator_responses = deque()
+        self.tokenfactory_module_state_responses = deque()
+
+    async def Params(self, request: token_factory_query_pb.QueryParamsRequest, context=None, metadata=None):
+        return self.params_responses.pop()
+
+    async def DenomAuthorityMetadata(
+        self, request: token_factory_query_pb.QueryDenomAuthorityMetadataRequest, context=None, metadata=None
+    ):
+        return self.denom_authority_metadata_responses.pop()
+
+    async def DenomsFromCreator(
+        self, request: token_factory_query_pb.QueryDenomsFromCreatorRequest, context=None, metadata=None
+    ):
+        return self.denoms_from_creator_responses.pop()
+
+    async def TokenfactoryModuleState(
+        self, request: token_factory_query_pb.QueryModuleStateRequest, context=None, metadata=None
+    ):
+        return self.tokenfactory_module_state_responses.pop()

tests/client/chain/grpc/test_chain_grpc_token_factory_api.py

@@ -0,0 +1,159 @@
+import grpc
+import pytest
+
+from pyinjective.client.chain.grpc.chain_grpc_token_factory_api import ChainGrpcTokenFactoryApi
+from pyinjective.core.network import Network
+from pyinjective.proto.cosmos.base.v1beta1 import coin_pb2 as coin_pb
+from pyinjective.proto.injective.tokenfactory.v1beta1 import (
+    authorityMetadata_pb2 as token_factory_authority_metadata_pb,
+    genesis_pb2 as token_factory_genesis_pb,
+    params_pb2 as token_factory_params_pb,
+    query_pb2 as token_factory_query_pb,
+)
+from tests.client.chain.grpc.configurable_token_factory_query_servicer import ConfigurableTokenFactoryQueryServicer
+
+
+@pytest.fixture
+def token_factory_query_servicer():
+    return ConfigurableTokenFactoryQueryServicer()
+
+
+class TestChainGrpcTokenFactoryApi:
+    @pytest.mark.asyncio
+    async def test_fetch_module_params(
+        self,
+        token_factory_query_servicer,
+    ):
+        coin = coin_pb.Coin(denom="inj", amount="988987297011197594664")
+        params = token_factory_params_pb.Params(
+            denom_creation_fee=[coin],
+        )
+        token_factory_query_servicer.params_responses.append(token_factory_query_pb.QueryParamsResponse(params=params))
+
+        network = Network.devnet()
+        channel = grpc.aio.insecure_channel(network.grpc_endpoint)
+
+        api = ChainGrpcTokenFactoryApi(channel=channel, metadata_provider=lambda: self._dummy_metadata_provider())
+        api._query_stub = token_factory_query_servicer
+
+        module_params = await api.fetch_module_params()
+        expected_params = {
+            "params": {
+                "denomCreationFee": [
+                    {"denom": coin.denom, "amount": coin.amount},
+                ],
+            }
+        }
+
+        assert module_params == expected_params
+
+    @pytest.mark.asyncio
+    async def test_fetch_denom_authority_metadata(
+        self,
+        token_factory_query_servicer,
+    ):
+        authority_metadata = token_factory_authority_metadata_pb.DenomAuthorityMetadata(
+            admin="inj1ady3s7whq30l4fx8sj3x6muv5mx4dfdlcpv8n7",
+        )
+        token_factory_query_servicer.denom_authority_metadata_responses.append(
+            token_factory_query_pb.QueryDenomAuthorityMetadataResponse(
+                authority_metadata=authority_metadata,
+            )
+        )
+
+        network = Network.devnet()
+        channel = grpc.aio.insecure_channel(network.grpc_endpoint)
+
+        api = ChainGrpcTokenFactoryApi(channel=channel, metadata_provider=lambda: self._dummy_metadata_provider())
+        api._query_stub = token_factory_query_servicer
+
+        metadata = await api.fetch_denom_authority_metadata(
+            creator=authority_metadata.admin,
+            sub_denom="inj",
+        )
+        expected_metadata = {
+            "authorityMetadata": {
+                "admin": authority_metadata.admin,
+            },
+        }
+
+        assert metadata == expected_metadata
+
+    @pytest.mark.asyncio
+    async def test_fetch_denoms_from_creator(
+        self,
+        token_factory_query_servicer,
+    ):
+        denom = "factory/inj17vytdwqczqz72j65saukplrktd4gyfme5agf6c/weth"
+        token_factory_query_servicer.denoms_from_creator_responses.append(
+            token_factory_query_pb.QueryDenomsFromCreatorResponse(denoms=[denom])
+        )
+
+        network = Network.devnet()
+        channel = grpc.aio.insecure_channel(network.grpc_endpoint)
+
+        api = ChainGrpcTokenFactoryApi(channel=channel, metadata_provider=lambda: self._dummy_metadata_provider())
+        api._query_stub = token_factory_query_servicer
+
+        denoms = await api.fetch_denoms_from_creator(creator="inj1ady3s7whq30l4fx8sj3x6muv5mx4dfdlcpv8n7")
+        expected_denoms = {"denoms": [denom]}
+
+        assert denoms == expected_denoms
+
+    @pytest.mark.asyncio
+    async def test_fetch_tokenfactory_module_state(
+        self,
+        token_factory_query_servicer,
+    ):
+        coin = coin_pb.Coin(denom="inj", amount="988987297011197594664")
+        params = token_factory_params_pb.Params(
+            denom_creation_fee=[coin],
+        )
+        authority_metadata = token_factory_authority_metadata_pb.DenomAuthorityMetadata(
+            admin="inj1apmvarl2xyv6kecx2ukkeymddw3we4zkygjyc0",
+        )
+        genesis_denom = token_factory_genesis_pb.GenesisDenom(
+            denom="factory/inj1apmvarl2xyv6kecx2ukkeymddw3we4zkygjyc0/ninja",
+            authority_metadata=authority_metadata,
+            name="Dog Wif Nunchucks",
+            symbol="NINJA",
+        )
+        state = token_factory_genesis_pb.GenesisState(
+            params=params,
+            factory_denoms=[genesis_denom],
+        )
+        token_factory_query_servicer.tokenfactory_module_state_responses.append(
+            token_factory_query_pb.QueryModuleStateResponse(state=state)
+        )
+
+        network = Network.devnet()
+        channel = grpc.aio.insecure_channel(network.grpc_endpoint)
+
+        api = ChainGrpcTokenFactoryApi(channel=channel, metadata_provider=lambda: self._dummy_metadata_provider())
+        api._query_stub = token_factory_query_servicer
+
+        state = await api.fetch_tokenfactory_module_state()
+        expected_state = {
+            "state": {
+                "params": {
+                    "denomCreationFee": [
+                        {"denom": coin.denom, "amount": coin.amount},
+                    ],
+                },
+                "factoryDenoms": [
+                    {
+                        "denom": genesis_denom.denom,
+                        "authorityMetadata": {
+                            "admin": authority_metadata.admin,
+                        },
+                        "name": genesis_denom.name,
+                        "symbol": genesis_denom.symbol,
+                    }
+                ],
+            }
+        }
+
+        assert state == expected_state
+
+    async def _dummy_metadata_provider(self):
+        return None