Merge branch 'jac/dataclass'

Author: jac18281828

.devcontainer/devcontainer.json

@@ -24,7 +24,7 @@
 	// 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",
 
-	// "runArgs": [  ],
+	"runArgs": [ "-e", "BN_API_KEY=EXAMPLEAPIKEY" ],
 
 	// 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" ],

blocknative/__init__.py

@@ -1 +1 @@
-__version__ = '0.2.2'
+__version__ = '0.2.3'

blocknative/stream.py

@@ -32,6 +32,8 @@
 
 BN_BASE_URL = 'wss://api.blocknative.com/v0'
 BN_ETHEREUM = 'ethereum'
+BN_ETHEREUM_ID = 1
+BN_STREAM_CLASS_VERSION = '1.1'
 
 Callback = Callable[[dict, Callable], None]
 
@@ -84,23 +86,23 @@ def as_dict(self) -> dict:
         }
 
 
-@dataclass
 class Stream:
     """Stream class used to connect to Blocknative's WebSocket API."""
 
     api_key: str
     blockchain: str = BN_ETHEREUM
-    network_id: int = 1
-    version: str = '1'
+    network_id: int = BN_ETHEREUM_ID
+    version: str = BN_STREAM_CLASS_VERSION
     global_filters: List[dict] = None
     valid_session: bool = True
-    _ws: WebSocketConnection = field(default=None, init=False)
-    _message_queue: Queue = field(default=Queue(), init=False)
+    _ws: WebSocketConnection = None
+    _message_queue: Queue = Queue()
+    _subscription_registry: Mapping[str, Subscription] = {}
 
-    # Registry of active subscriptions.
-    _subscription_registry: Mapping[str, Subscription] = field(
-        default_factory=dict, init=False
-    )
+    def __init__(self, api_key: str, blockchain: str = BN_ETHEREUM, network_id: int = BN_ETHEREUM_ID):
+        self.api_key=api_key
+        self.blockchain=blockchain
+        self.network_id=network_id
 
     def subscribe_address(
         self,
@@ -156,7 +158,7 @@ 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, base_url:str = BN_BASE_URL):
+    def connect(self, base_url: str = BN_BASE_URL):
         """Initializes the connection to the WebSocket server."""
         try:
             return trio.run(self._connect, base_url)
@@ -170,8 +172,9 @@ def send_message(self, message: str):
         Args:
             message: The message to send.
         """
-        logging.debug('Sending: {}' % message)
         self._message_queue.put(message)
+        logging.debug('Sending: %s', message)
+
 
     async def _message_dispatcher(self):
         """In a loop: Polls send message queue for latest messages to send to server.

examples/subscribe.py

@@ -16,8 +16,9 @@ async def txn_handler(txn, unsubscribe):
             with open(apikeyfile, 'r') as apikey:
                 keystring = apikey.readline().rstrip().lstrip()
                 stream = BNStream(keystring)
-                stream.subscribe_address(monitor_address, txn_handler)
+                filter = {'network': 'main'}
+                stream.subscribe_address(monitor_address, txn_handler, filters=[filter])
                 stream.connect()
     except Exception as e:
-        print('API Failed: %s' % str(e))
+        logging.error('API Failed: %s', str(e))
         traceback.print_exc(e)

examples/subscribe_bsc.py

@@ -0,0 +1,25 @@
+from blocknative.stream import Stream as BNStream
+import json,sys,traceback,logging
+
+NETWORK_ID = 56
+
+async def txn_handler(txn, unsubscribe):
+    # Output the transaction data to the console
+    print(json.dumps(txn, indent=4))
+
+if __name__ == '__main__':
+        if len(sys.argv) == 1:
+            print('%s apikey' % sys.argv[0])
+        else:
+            try:
+                logging.basicConfig(level=logging.INFO)            
+                apikeyfile = sys.argv[1]
+                with open(apikeyfile, 'r') as apikey:
+                    keystring = apikey.readline().rstrip().lstrip()
+                    stream = BNStream(keystring, network_id=NETWORK_ID)
+                    pancakeswap_v2_address = '0x10ed43c718714eb63d5aa57b78b54704e256024e'
+                    stream.subscribe_address(pancakeswap_v2_address, txn_handler)
+                    stream.connect()
+            except Exception as e:
+                logging.error('API Failed: %s', str(e))
+                traceback.print_exc(e)