Merge branch 'jac/flatten_event'

.devcontainer/devcontainer.json

@@ -16,16 +16,15 @@
 	// Add the IDs of extensions you want installed when the container is created.
 	"extensions": [
 		"ms-python.python"
-	]
+	],
 
 	// Use 'forwardPorts' to make a list of ports inside the container available locally.
 	// "forwardPorts": [],
 
 	// Uncomment the next line to run commands after the container is created - for example installing curl.
 	// "postCreateCommand": "apt-get update && apt-get install -y curl",
 
-	// Uncomment when using a ptrace-based debugger like C++, Go, and Rust
-	// "runArgs": [ "--cap-add=SYS_PTRACE", "--security-opt", "seccomp=unconfined" ],
+	// "runArgs": [  ],
 
 	// Uncomment to use the Docker CLI from inside the container. See https://aka.ms/vscode-remote/samples/docker-from-docker.
 	// "mounts": [ "source=/var/run/docker.sock,target=/var/run/docker.sock,type=bind" ],

.github/workflows/publish-to-test-pypi.yml

@@ -14,7 +14,23 @@ jobs:
         uses: actions/setup-python@v1
         with:
           python-version: 3.9
-      - name: Install pypa/build
+      - name: upgrade pip
+        run: >-
+          python -m
+          pip install --upgrade pip
+      - name: py requirements
+        run: >-
+          python -m
+          pip install -r requirements.txt          
+      - name: py compile
+        run: >-
+          python -m
+          py_compile blocknative/*.py
+      - name: python unittest
+        run: >-
+          python -m
+          unittest discover -s tests -p '*test.py'
+      - name: build
         run: >-
           python -m
           pip install

Dockerfile

@@ -20,8 +20,9 @@ COPY tests tests/
 COPY setup.py .
 RUN pip3 install --upgrade pip
 RUN pip3 install -r requirements.txt
+RUN pip3 install --upgrade autopep8
 RUN python3 setup.py install
 ENV PYTHONPATH=.
 RUN python3 -m py_compile blocknative/*.py
-RUN python3 -m unittest discover -s tests
+RUN python3 -m unittest discover -s tests -p '*test.py'
 CMD echo Python SDK

blocknative/__init__.py

@@ -1 +1 @@
-__version__ = '0.1.8'
+__version__ = '0.2.0'

blocknative/stream.py

@@ -7,8 +7,8 @@
 from dataclasses import dataclass, field
 from queue import Queue, Empty
 from typing import List, Mapping, Callable, Union
-import types
 import trio
+import logging
 from trio_websocket import (
     open_websocket_url,
     ConnectionClosed,
@@ -24,13 +24,16 @@
     SubscriptionType,
     to_camel_case,
 )
+from blocknative import __version__ as API_VERSION
 
 PING_INTERVAL = 15
 PING_TIMEOUT = 10
 MESSAGE_SEND_INTERVAL = 0.021  # 21ms
 
-Callback = Callable[[dict, Callable], None]
+BN_BASE_URL = 'wss://api.blocknative.com/v0'
+BN_ETHEREUM = 'ethereum'
 
+Callback = Callable[[dict, Callable], None]
 
 @dataclass
 class Subscription:
@@ -85,18 +88,12 @@ def as_dict(self) -> dict:
 class Stream:
     """Stream class used to connect to Blocknative's WebSocket API."""
 
-    # - Public instance variables -
-
     api_key: str
-    base_url: str = 'wss://api.blocknative.com/v0'
-    blockchain: str = 'ethereum'
+    blockchain: str = BN_ETHEREUM
     network_id: int = 1
     version: str = '1'
     global_filters: List[dict] = None
     valid_session: bool = True
-
-    # - Private instance variables -
-
     _ws: WebSocketConnection = field(default=None, init=False)
     _message_queue: Queue = field(default=Queue(), init=False)
 
@@ -128,7 +125,7 @@ async def txn_handler(txn)
             stream.subscribe('0x7a250d5630b4cf539739df2c5dacb4c659f2488d', txn_handler)
         """
 
-        if self.blockchain == 'ethereum':
+        if self.blockchain == BN_ETHEREUM:
             address = address.lower()
 
         # Add this subscription to the registry
@@ -159,10 +156,10 @@ def subscribe_txn(self, tx_hash: str, callback: Callback, status: str = 'sent'):
         if self._is_connected():
             self._send_txn_watch_message(tx_hash, status)
 
-    def connect(self):
+    def connect(self, base_url:str = BN_BASE_URL):
         """Initializes the connection to the WebSocket server."""
         try:
-            return trio.run(self._connect)
+            return trio.run(self._connect, base_url)
         except KeyboardInterrupt:
             print('keyboard interrupt')
             return None
@@ -173,6 +170,7 @@ def send_message(self, message: str):
         Args:
             message: The message to send.
         """
+        logging.debug('Sending: {}' % message)
         self._message_queue.put(message)
 
     async def _message_dispatcher(self):
@@ -220,34 +218,29 @@ async def _message_handler(self, message: dict):
         # Raises an exception if the status of the message is an error
         raise_error_on_status(message)
 
-        if 'event' in message and 'transaction' in message['event']:
+        if 'event' in message: 
+            event = message['event']
             # Ignore server echo and unsubscribe messages
-            if is_server_echo(message['event']['eventCode']):
+            if is_server_echo(event['eventCode']):
                 return
 
-            # Checks if the messsage is for a transaction subscription
-            if subscription_type(message) == SubscriptionType.TRANSACTION:
-
-                # Find the matching subscription and run it's callback
-                if (
-                    message['event']['transaction']['hash']
-                    in self._subscription_registry
-                ):
-                    await self._subscription_registry[
-                        message['event']['transaction']['hash']
-                    ].callback(message['event']['transaction'])
-
-            # Checks if the messsage is for an address subscription
-            elif subscription_type(message) == SubscriptionType.ADDRESS:
-                watched_address = message['event']['transaction']['watchedAddress']
-                if watched_address in self._subscription_registry and watched_address is not None:
+            if 'transaction' in event:
+                event_transaction = event['transaction']
+                # Checks if the messsage is for a transaction subscription
+                if subscription_type(message) == SubscriptionType.TRANSACTION:
                     # Find the matching subscription and run it's callback
-                    if 'transaction' in message['event']:
-                        transaction = message['event']['transaction']
-                        await self._subscription_registry[watched_address].callback(
-                            transaction,
-                            (lambda: self.unsubscribe(watched_address)),
-                        )
+                    transaction_hash = event_transaction['hash']
+                    if transaction_hash in self._subscription_registry:
+                        transaction = self._flatten_event_to_transaction(event)
+                        await self._subscription_registry[transaction_hash].callback(transaction)
+
+                # Checks if the messsage is for an address subscription
+                elif subscription_type(message) == SubscriptionType.ADDRESS:
+                    watched_address = event_transaction['watchedAddress']
+                    if watched_address in self._subscription_registry and watched_address is not None:
+                        # Find the matching subscription and run it's callback
+                        transaction = self._flatten_event_to_transaction(event)
+                        await self._subscription_registry[watched_address].callback(transaction,(lambda: self.unsubscribe(watched_address)))
 
     def unsubscribe(self, watched_address):
         # remove this subscription from the registry so that we don't execute the callback
@@ -284,7 +277,7 @@ async def _heartbeat(self):
                 await self._ws.ping()
             await trio.sleep(PING_INTERVAL)
 
-    async def _handle_connection(self):
+    async def _handle_connection(self, base_url:str):
         """Handles the setup once the websocket connection is established, as well as,
         handles reconnect if the websocket closes for any reason.
 
@@ -315,14 +308,16 @@ async def _handle_connection(self):
                 nursery.start_soon(self._message_dispatcher)
         except ConnectionClosed as cc:
             # If server times the connection out or drops, reconnect
-            await self._connect()
+            await trio.sleep(0.5)
+            await self._connect(base_url)
 
-    async def _connect(self):
+    async def _connect(self, base_url):
         try:
-            async with open_websocket_url(self.base_url) as ws:
+            async with open_websocket_url(base_url) as ws:
                 self._ws = ws
-                await self._handle_connection()
+                await self._handle_connection(base_url)
         except HandshakeError as e:
+            logging.exception('Handshake failed')
             return False
 
     def _is_connected(self) -> bool:
@@ -398,7 +393,7 @@ def _build_payload(
         return {
             'timeStamp': datetime.now().isoformat(),
             'dappId': self.api_key,
-            'version': self.version,
+            'version': API_VERSION,
             'blockchain': {
                 'system': self.blockchain,
                 'network': network_id_to_name(self.network_id),
@@ -413,3 +408,25 @@ def _queue_init_message(self):
         self.send_message(
             self._build_payload(category_code='initialize', event_code='checkDappId')
         )
+
+    def _flatten_event_to_transaction(self, event:dict):
+        transaction = {}
+        eventcopy = dict(event)
+        del eventcopy['dappId']
+        if 'transaction' in eventcopy:
+            txn = eventcopy['transaction']
+            for k in txn.keys():
+                transaction[k] = txn[k]
+            del eventcopy['transaction']
+        if 'blockchain' in eventcopy:
+            bc = eventcopy['blockchain']
+            for k in bc.keys():
+                transaction[k] = bc[k]
+            del eventcopy['blockchain']
+        if 'contractCall' in eventcopy:
+            transaction['contractCall'] = eventcopy['contractCall']
+            del eventcopy['contractCall']
+        for k in eventcopy:
+            if not isinstance(k, dict) and not isinstance(k, list):
+                transaction[k] = eventcopy[k]
+        return transaction

examples/confirm_n.py

@@ -1,5 +1,6 @@
 from blocknative.stream import Stream as BNStream
-import json,sys,traceback,types
+import json,sys,traceback
+import logging
 
 monitor_address = '0x7a250d5630b4cf539739df2c5dacb4c659f2488d'
 
@@ -11,23 +12,26 @@ def __init__(self):
 
     async def on_transaction(self, txn, unsubscribe):
         print(json.dumps(txn, indent=4))
-        if 'status' in txn and txn['status'] == 'confirmed':
+        if txn['status'] == 'confirmed':
             self.transaction_count -= 1
-            if self.transaction_count < 0:
+            if self.transaction_count < 1:
                 unsubscribe()
 
 if __name__ == '__main__':
     try:
         if len(sys.argv) == 1:
-            print('{} apikey' % sys.argv[0])
+            print('%s apikey' % sys.argv[0])
         else:
+            logging.basicConfig(level=logging.INFO)
+
             apikeyfile = sys.argv[1]
             with open(apikeyfile, 'r') as apikey:
                 keystring = apikey.readline().rstrip().lstrip()
                 stream = BNStream(keystring)
                 printer = Printer()
-                stream.subscribe_address(monitor_address, (lambda txn, callback: printer.on_transaction(txn, callback)))
+                server_filter = [{'status':'confirmed'}]
+                stream.subscribe_address(monitor_address, (lambda txn, callback: printer.on_transaction(txn, callback)), filters=server_filter)
                 stream.connect()
     except Exception as e:
-        print('API Failed: ' + str(e))
+        print('API Failed: %s' % str(e))
         traceback.print_exc(e)

examples/subscribe.py

@@ -1,5 +1,5 @@
 from blocknative.stream import Stream as BNStream
-import json,sys,traceback
+import json,sys,traceback,logging
 
 monitor_address = '0x7a250d5630b4cf539739df2c5dacb4c659f2488d'
 
@@ -9,14 +9,15 @@ async def txn_handler(txn, unsubscribe):
 if __name__ == '__main__':
     try:
         if len(sys.argv) == 1:
-            print('{} apikey' % sys.argv[0])
+            print('%s apikey' % sys.argv[0])
         else:
+            logging.basicConfig(level=logging.INFO)            
             apikeyfile = sys.argv[1]
             with open(apikeyfile, 'r') as apikey:
                 keystring = apikey.readline().rstrip().lstrip()
                 stream = BNStream(keystring)
                 stream.subscribe_address(monitor_address, txn_handler)
                 stream.connect()
     except Exception as e:
-        print('API Failed: ' + str(e))
-        traceback.print_exc(e)
+        print('API Failed: %s' % str(e))
+        traceback.print_exc(e)

examples/unsubscribe_confirmed.py

@@ -1,5 +1,5 @@
 from blocknative.stream import Stream as BNStream
-import json,sys,traceback
+import json,sys,traceback,logging
 
 monitor_address = '0x7a250d5630b4cf539739df2c5dacb4c659f2488d'
 
@@ -11,14 +11,15 @@ async def txn_handler(txn, unsubscribe):
 if __name__ == '__main__':
     try:
         if len(sys.argv) == 1:
-            print('{} apikey' % sys.argv[0])
+            print('%s apikey' % sys.argv[0])
         else:
+            logging.basicConfig(level=logging.INFO)            
             apikeyfile = sys.argv[1]
             with open(apikeyfile, 'r') as apikey:
                 keystring = apikey.readline().rstrip().lstrip()
                 stream = BNStream(keystring)
                 stream.subscribe_address(monitor_address, txn_handler)
                 stream.connect()
     except Exception as e:
-        print('API Failed: ' + str(e))
-        traceback.print_exc(e)
+        print('API Failed: %s' % str(e))
+        traceback.print_exc(e)