(feat) Added low level API component for exchange explorer streams. Added unit tests for the new functionality. Included new functions in AsyncClient using the low level API components, and marked the old functions as deprecated. Updated the example scripts.

abel · abel · commit d65cb2e7f51c · 2023-12-01T13:18:02.000-03:00
diff --git a/examples/exchange_client/explorer_rpc/6_StreamTxs.py b/examples/exchange_client/explorer_rpc/6_StreamTxs.py
@@ -1,16 +1,39 @@
 import asyncio
+from typing import Any, Dict
+
+from grpc import RpcError
 
 from pyinjective.async_client import AsyncClient
 from pyinjective.core.network import Network
 
 
+async def tx_event_processor(event: Dict[str, Any]):
+    print(event)
+
+
+def stream_error_processor(exception: RpcError):
+    print(f"There was an error listening to txs updates ({exception})")
+
+
+def stream_closed_processor():
+    print("The txs updates stream has been closed")
+
+
 async def main() -> None:
     # select network: local, testnet, mainnet
     network = Network.testnet()
     client = AsyncClient(network)
-    stream_txs = await client.stream_txs()
-    async for tx in stream_txs:
-        print(tx)
+
+    task = asyncio.get_event_loop().create_task(
+        client.listen_txs_updates(
+            callback=tx_event_processor,
+            on_end_callback=stream_closed_processor,
+            on_status_callback=stream_error_processor,
+        )
+    )
+
+    await asyncio.sleep(delay=60)
+    task.cancel()
 
 
 if __name__ == "__main__":
diff --git a/examples/exchange_client/explorer_rpc/7_StreamBlocks.py b/examples/exchange_client/explorer_rpc/7_StreamBlocks.py
@@ -1,16 +1,39 @@
 import asyncio
+from typing import Any, Dict
+
+from grpc import RpcError
 
 from pyinjective.async_client import AsyncClient
 from pyinjective.core.network import Network
 
 
+async def block_event_processor(event: Dict[str, Any]):
+    print(event)
+
+
+def stream_error_processor(exception: RpcError):
+    print(f"There was an error listening to blocks updates ({exception})")
+
+
+def stream_closed_processor():
+    print("The blocks updates stream has been closed")
+
+
 async def main() -> None:
     # select network: local, testnet, mainnet
     network = Network.testnet()
     client = AsyncClient(network)
-    stream_blocks = await client.stream_blocks()
-    async for block in stream_blocks:
-        print(block)
+
+    task = asyncio.get_event_loop().create_task(
+        client.listen_blocks_updates(
+            callback=block_event_processor,
+            on_end_callback=stream_closed_processor,
+            on_status_callback=stream_error_processor,
+        )
+    )
+
+    await asyncio.sleep(delay=60)
+    task.cancel()
 
 
 if __name__ == "__main__":
diff --git a/pyinjective/async_client.py b/pyinjective/async_client.py
@@ -24,6 +24,7 @@
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_account_stream import IndexerGrpcAccountStream
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_auction_stream import IndexerGrpcAuctionStream
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_derivative_stream import IndexerGrpcDerivativeStream
+from pyinjective.client.indexer.grpc_stream.indexer_grpc_explorer_stream import IndexerGrpcExplorerStream
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_meta_stream import IndexerGrpcMetaStream
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_oracle_stream import IndexerGrpcOracleStream
 from pyinjective.client.indexer.grpc_stream.indexer_grpc_portfolio_stream import IndexerGrpcPortfolioStream
@@ -278,6 +279,12 @@ def __init__(
                 metadata_query_provider=self._explorer_cookie_metadata_requestor
             ),
         )
+        self.exchange_explorer_stream_api = IndexerGrpcExplorerStream(
+            channel=self.explorer_channel,
+            metadata_provider=lambda: self.network.exchange_metadata(
+                metadata_query_provider=self._explorer_cookie_metadata_requestor
+            ),
+        )
 
     async def all_tokens(self) -> Dict[str, Token]:
         if self._tokens is None:
@@ -729,13 +736,45 @@ async def fetch_txs(
         )
 
     async def stream_txs(self):
+        """
+        This method is deprecated and will be removed soon. Please use `listen_txs_updates` instead
+        """
+        warn("This method is deprecated. Use listen_txs_updates instead", DeprecationWarning, stacklevel=2)
         req = explorer_rpc_pb.StreamTxsRequest()
         return self.stubExplorer.StreamTxs(req)
 
+    async def listen_txs_updates(
+        self,
+        callback: Callable,
+        on_end_callback: Optional[Callable] = None,
+        on_status_callback: Optional[Callable] = None,
+    ):
+        await self.exchange_explorer_stream_api.stream_txs(
+            callback=callback,
+            on_end_callback=on_end_callback,
+            on_status_callback=on_status_callback,
+        )
+
     async def stream_blocks(self):
+        """
+        This method is deprecated and will be removed soon. Please use `listen_blocks_updates` instead
+        """
+        warn("This method is deprecated. Use listen_blocks_updates instead", DeprecationWarning, stacklevel=2)
         req = explorer_rpc_pb.StreamBlocksRequest()
         return self.stubExplorer.StreamBlocks(req)
 
+    async def listen_blocks_updates(
+        self,
+        callback: Callable,
+        on_end_callback: Optional[Callable] = None,
+        on_status_callback: Optional[Callable] = None,
+    ):
+        await self.exchange_explorer_stream_api.stream_blocks(
+            callback=callback,
+            on_end_callback=on_end_callback,
+            on_status_callback=on_status_callback,
+        )
+
     async def get_peggy_deposits(self, **kwargs):
         """
         This method is deprecated and will be removed soon. Please use `fetch_peggy_deposit_txs` instead
diff --git a/pyinjective/client/indexer/grpc_stream/indexer_grpc_explorer_stream.py b/pyinjective/client/indexer/grpc_stream/indexer_grpc_explorer_stream.py
@@ -0,0 +1,47 @@
+from typing import Callable, Optional
+
+from grpc.aio import Channel
+
+from pyinjective.proto.exchange import (
+    injective_explorer_rpc_pb2 as exchange_explorer_pb,
+    injective_explorer_rpc_pb2_grpc as exchange_explorer_grpc,
+)
+from pyinjective.utils.grpc_api_stream_assistant import GrpcApiStreamAssistant
+
+
+class IndexerGrpcExplorerStream:
+    def __init__(self, channel: Channel, metadata_provider: Callable):
+        self._stub = self._stub = exchange_explorer_grpc.InjectiveExplorerRPCStub(channel)
+        self._assistant = GrpcApiStreamAssistant(metadata_provider=metadata_provider)
+
+    async def stream_txs(
+        self,
+        callback: Callable,
+        on_end_callback: Optional[Callable] = None,
+        on_status_callback: Optional[Callable] = None,
+    ):
+        request = exchange_explorer_pb.StreamTxsRequest()
+
+        await self._assistant.listen_stream(
+            call=self._stub.StreamTxs,
+            request=request,
+            callback=callback,
+            on_end_callback=on_end_callback,
+            on_status_callback=on_status_callback,
+        )
+
+    async def stream_blocks(
+        self,
+        callback: Callable,
+        on_end_callback: Optional[Callable] = None,
+        on_status_callback: Optional[Callable] = None,
+    ):
+        request = exchange_explorer_pb.StreamBlocksRequest()
+
+        await self._assistant.listen_stream(
+            call=self._stub.StreamBlocks,
+            request=request,
+            callback=callback,
+            on_end_callback=on_end_callback,
+            on_status_callback=on_status_callback,
+        )
diff --git a/tests/client/indexer/configurable_explorer_query_servicer.py b/tests/client/indexer/configurable_explorer_query_servicer.py
@@ -29,6 +29,9 @@ def __init__(self):
         self.relayers_responses = deque()
         self.bank_transfers_responses = deque()
 
+        self.stream_txs_responses = deque()
+        self.stream_blocks_responses = deque()
+
     async def GetAccountTxs(self, request: exchange_explorer_pb.GetAccountTxsRequest, context=None, metadata=None):
         return self.account_txs_responses.pop()
 
@@ -99,3 +102,11 @@ async def GetBankTransfers(
         self, request: exchange_explorer_pb.GetBankTransfersRequest, context=None, metadata=None
     ):
         return self.bank_transfers_responses.pop()
+
+    async def StreamTxs(self, request: exchange_explorer_pb.StreamTxsRequest, context=None, metadata=None):
+        for event in self.stream_txs_responses:
+            yield event
+
+    async def StreamBlocks(self, request: exchange_explorer_pb.StreamBlocksRequest, context=None, metadata=None):
+        for event in self.stream_blocks_responses:
+            yield event
diff --git a/tests/client/indexer/stream_grpc/test_indexer_grpc_explorer_stream.py b/tests/client/indexer/stream_grpc/test_indexer_grpc_explorer_stream.py
@@ -0,0 +1,138 @@
+import asyncio
+
+import grpc
+import pytest
+
+from pyinjective.client.indexer.grpc_stream.indexer_grpc_explorer_stream import IndexerGrpcExplorerStream
+from pyinjective.core.network import Network
+from pyinjective.proto.exchange import injective_explorer_rpc_pb2 as exchange_explorer_pb
+from tests.client.indexer.configurable_explorer_query_servicer import ConfigurableExplorerQueryServicer
+
+
+@pytest.fixture
+def explorer_servicer():
+    return ConfigurableExplorerQueryServicer()
+
+
+class TestIndexerGrpcAuctionStream:
+    @pytest.mark.asyncio
+    async def test_stream_txs(
+        self,
+        explorer_servicer,
+    ):
+        code = 5
+        claim_id = 100
+        tx_data = exchange_explorer_pb.StreamTxsResponse(
+            id="test id",
+            block_number=18138926,
+            block_timestamp="2023-11-07 23:19:55.371 +0000 UTC",
+            hash="0x3790ade2bea6c8605851ec89fa968adf2a2037a5ecac11ca95e99260508a3b7e",
+            codespace="test codespace",
+            messages='[{"type":"/cosmos.bank.v1beta1.MsgSend",'
+            '"value":{"from_address":"inj1phd706jqzd9wznkk5hgsfkrc8jqxv0kmlj0kex",'
+            '"to_address":"inj1d6qx83nhx3a3gx7e654x4su8hur5s83u84h2xc",'
+            '"amount":[{"denom":"factory/inj17vytdwqczqz72j65saukplrktd4gyfme5agf6c/weth",'
+            '"amount":"100000000000000000"}]}}]',
+            tx_number=221429,
+            error_log="",
+            code=code,
+            claim_ids=[claim_id],
+        )
+
+        explorer_servicer.stream_txs_responses.append(tx_data)
+
+        network = Network.devnet()
+        channel = grpc.aio.insecure_channel(network.grpc_exchange_endpoint)
+
+        api = IndexerGrpcExplorerStream(channel=channel, metadata_provider=lambda: self._dummy_metadata_provider())
+        api._stub = explorer_servicer
+
+        txs_updates = asyncio.Queue()
+        end_event = asyncio.Event()
+
+        callback = lambda update: txs_updates.put_nowait(update)
+        error_callback = lambda exception: pytest.fail(str(exception))
+        end_callback = lambda: end_event.set()
+
+        asyncio.get_event_loop().create_task(
+            api.stream_txs(
+                callback=callback,
+                on_end_callback=end_callback,
+                on_status_callback=error_callback,
+            )
+        )
+        expected_update = {
+            "id": tx_data.id,
+            "blockNumber": str(tx_data.block_number),
+            "blockTimestamp": tx_data.block_timestamp,
+            "hash": tx_data.hash,
+            "codespace": tx_data.codespace,
+            "messages": tx_data.messages,
+            "txNumber": str(tx_data.tx_number),
+            "errorLog": tx_data.error_log,
+            "code": tx_data.code,
+            "claimIds": [str(claim_id)],
+        }
+
+        first_update = await asyncio.wait_for(txs_updates.get(), timeout=1)
+
+        assert first_update == expected_update
+        assert end_event.is_set()
+
+    @pytest.mark.asyncio
+    async def test_stream_blocks(
+        self,
+        explorer_servicer,
+    ):
+        block_info = exchange_explorer_pb.StreamBlocksResponse(
+            height=19034578,
+            proposer="injvalcons18x63wcw5hjxlf535lgn4qy20yer7mm0qedu0la",
+            moniker="InjectiveNode1",
+            block_hash="0x7f7bfe8caaa0eed042315d1447ef1ed726a80f5da23fdbe6831fc66775197db1",
+            parent_hash="0x44287ba5fad21d0109a3ec6f19d447580763e5a709e5a5ceb767174e99ae3bd8",
+            num_pre_commits=20,
+            num_txs=4,
+            timestamp="2023-11-29 20:23:33.842 +0000 UTC",
+        )
+
+        explorer_servicer.stream_blocks_responses.append(block_info)
+
+        network = Network.devnet()
+        channel = grpc.aio.insecure_channel(network.grpc_exchange_endpoint)
+
+        api = IndexerGrpcExplorerStream(channel=channel, metadata_provider=lambda: self._dummy_metadata_provider())
+        api._stub = explorer_servicer
+
+        blocks_updates = asyncio.Queue()
+        end_event = asyncio.Event()
+
+        callback = lambda update: blocks_updates.put_nowait(update)
+        error_callback = lambda exception: pytest.fail(str(exception))
+        end_callback = lambda: end_event.set()
+
+        asyncio.get_event_loop().create_task(
+            api.stream_blocks(
+                callback=callback,
+                on_end_callback=end_callback,
+                on_status_callback=error_callback,
+            )
+        )
+        expected_update = {
+            "height": str(block_info.height),
+            "proposer": block_info.proposer,
+            "moniker": block_info.moniker,
+            "blockHash": block_info.block_hash,
+            "parentHash": block_info.parent_hash,
+            "numPreCommits": str(block_info.num_pre_commits),
+            "numTxs": str(block_info.num_txs),
+            "txs": [],
+            "timestamp": block_info.timestamp,
+        }
+
+        first_update = await asyncio.wait_for(blocks_updates.get(), timeout=1)
+
+        assert first_update == expected_update
+        assert end_event.is_set()
+
+    async def _dummy_metadata_provider(self):
+        return None
diff --git a/tests/test_async_client_deprecation_warnings.py b/tests/test_async_client_deprecation_warnings.py
