Support websockets API

pyproject.toml

@@ -20,7 +20,8 @@ dependencies = [
     "aiohttp>=3.8.4,<4.0.0",
     "pycryptodome>=3.17,<4.0",
     "crypto-cpp-py==1.4.5",
-    "eth-keyfile>=0.8.1,<1.0.0"
+    "eth-keyfile>=0.8.1,<1.0.0",
+    "websockets>=15.0.1,<16.0.0",
 ]
 
 [project.optional-dependencies]

starknet_py/conftest.py

@@ -7,6 +7,7 @@
     "starknet_py.tests.e2e.fixtures.contracts_v1",
     "starknet_py.tests.e2e.fixtures.misc",
     "starknet_py.tests.e2e.fixtures.devnet",
+    "starknet_py.tests.e2e.fixtures.devnet_ws",
     "starknet_py.tests.e2e.fixtures.constants",
     "starknet_py.tests.e2e.client.fixtures.transactions",
     "starknet_py.tests.e2e.client.fixtures.prepare_network",

starknet_py/net/client_models.py

@@ -33,6 +33,7 @@
 
 Hash = Union[int, str]
 Tag = Literal["pending", "latest"]
+LatestTag = Literal["latest"]
 
 
 @dataclass

starknet_py/net/client_utils.py

@@ -1,5 +1,5 @@
 import re
-from typing import Dict, Union, cast
+from typing import Dict, Optional, Union, cast
 
 from typing_extensions import get_args
 
@@ -86,3 +86,35 @@ def _create_broadcasted_txn(transaction: AccountTransaction) -> dict:
         Dict,
         BroadcastedTransactionSchema().dump(obj=transaction),
     )
+
+
+def get_block_identifier(
+    block_hash: Optional[Union[Hash, Tag]] = None,
+    block_number: Optional[Union[int, Tag]] = None,
+    default_tag: Optional[Tag] = "pending",
+) -> dict:
+    return {
+        "block_id": _get_raw_block_identifier(block_hash, block_number, default_tag)
+    }
+
+
+def _get_raw_block_identifier(
+    block_hash: Optional[Union[Hash, Tag]] = None,
+    block_number: Optional[Union[int, Tag]] = None,
+    default_tag: Optional[Tag] = "pending",
+) -> Union[dict, Hash, Tag, None]:
+    if block_hash is not None and block_number is not None:
+        raise ValueError(
+            "Arguments block_hash and block_number are mutually exclusive."
+        )
+
+    if block_hash in ("latest", "pending") or block_number in ("latest", "pending"):
+        return block_hash or block_number
+
+    if block_hash is not None:
+        return {"block_hash": _to_rpc_felt(block_hash)}
+
+    if block_number is not None:
+        return {"block_number": block_number}
+
+    return default_tag

starknet_py/net/full_node_client.py

@@ -41,10 +41,12 @@
 )
 from starknet_py.net.client_utils import (
     _create_broadcasted_txn,
+    _get_raw_block_identifier,
     _is_valid_eth_address,
     _to_rpc_felt,
     _to_storage_key,
     encode_l1_message,
+    get_block_identifier,
 )
 from starknet_py.net.executable_models import CasmClass
 from starknet_py.net.http_client import RpcHttpClient
@@ -862,31 +864,3 @@ async def trace_block_transactions(
             List[BlockTransactionTrace],
             BlockTransactionTraceSchema().load(res, many=True),
         )
-
-
-def get_block_identifier(
-    block_hash: Optional[Union[Hash, Tag]] = None,
-    block_number: Optional[Union[int, Tag]] = None,
-) -> dict:
-    return {"block_id": _get_raw_block_identifier(block_hash, block_number)}
-
-
-def _get_raw_block_identifier(
-    block_hash: Optional[Union[Hash, Tag]] = None,
-    block_number: Optional[Union[int, Tag]] = None,
-) -> Union[dict, Hash, Tag, None]:
-    if block_hash is not None and block_number is not None:
-        raise ValueError(
-            "Arguments block_hash and block_number are mutually exclusive."
-        )
-
-    if block_hash in ("latest", "pending") or block_number in ("latest", "pending"):
-        return block_hash or block_number
-
-    if block_hash is not None:
-        return {"block_hash": _to_rpc_felt(block_hash)}
-
-    if block_number is not None:
-        return {"block_number": block_number}
-
-    return "pending"

starknet_py/net/schemas/rpc/block.py

@@ -2,6 +2,7 @@
 
 from starknet_py.net.client_models import (
     BlockHashAndNumber,
+    BlockHeader,
     BlockStateUpdate,
     ContractsNonce,
     DeclaredContractHash,
@@ -77,6 +78,10 @@ class BlockHeaderSchema(Schema):
     l1_da_mode = L1DAModeField(data_key="l1_da_mode", required=True)
     starknet_version = fields.String(data_key="starknet_version", required=True)
 
+    @post_load
+    def make_dataclass(self, data, **kwargs) -> BlockHeader:
+        return BlockHeader(**data)
+
 
 class BlockHashAndNumberSchema(Schema):
     block_hash = Felt(data_key="block_hash", required=True)