refactor: make examples accept addresses via CLI args and env vars

.github/workflows/examples.yml

@@ -26,6 +26,10 @@ jobs:
         python3 -m pip install .
 
     - name: Run all Examples
+      env:
+        JSONRPC_ADDRESS: ${{ secrets.EXAMPLE_JSONRPC_ADDRESS }}
+        GRPC_ADDRESS: ${{ secrets.EXAMPLE_GRPC_ADDRESS }}
+        ZMQ_ADDRESS: ${{ secrets.EXAMPLE_ZMQ_ADDRESS }}
       run: |
         for example in examples/*.py; do
           echo "=== Running $example"

examples/example_decode_block.py

@@ -10,24 +10,24 @@
 4. Print the block's details, including header, previous certificate, and transactions.
 """
 
+import argparse
 import asyncio
+import os
 import random
 
 from pactus.block import Block
 from pactus.encoding import Reader
 from pactus_jsonrpc.client import PactusOpenRPCClient
 
-TESTNET_RPC = "https://testnet1.pactus.org/jsonrpc"
 
-
-async def fetch_and_decode_block():
+async def decode_block(client_url: str):
     client = PactusOpenRPCClient(
         headers={},
         timeout=30,
-        client_url=TESTNET_RPC,
+        client_url=client_url,
     )
 
-    # print("Connecting to Pactus testnet...")
+    # print("Connecting to Pactus...")
     # info = await client.pactus.blockchain.get_blockchain_info()
     # latest_height = info["last_block_height"]
     # print(f"  Latest height: {latest_height}")
@@ -56,4 +56,18 @@ async def fetch_and_decode_block():
 
 
 if __name__ == "__main__":
-    asyncio.run(fetch_and_decode_block())
+    parser = argparse.ArgumentParser(
+        description="Fetch and decode a Pactus block via JSON-RPC."
+    )
+    parser.add_argument(
+        "--address",
+        default=os.environ.get("JSONRPC_ADDRESS"),
+        help="JSON-RPC server address (env: JSONRPC_ADDRESS)",
+    )
+    args = parser.parse_args()
+    if not args.address:
+        parser.error(
+            "No JSON-RPC address provided. Use --address or set JSONRPC_ADDRESS."
+        )
+
+    asyncio.run(decode_block(args.address))

examples/example_grpc_get_blockchain_info.py

@@ -1,21 +1,42 @@
-from pactus_grpc.blockchain_pb2_grpc import BlockchainStub
-from pactus_grpc.blockchain_pb2 import GetBlockchainInfoRequest
+#!/usr/bin/env python3
+"""
+Pactus SDK Example: Get Blockchain Info via gRPC
+=================================================
+
+This example demonstrates how to connect to a Pactus node via gRPC
+and retrieve blockchain information.
+"""
+
+import argparse
+import os
+
 import grpc
 
+from pactus_grpc.blockchain_pb2 import GetBlockchainInfoRequest
+from pactus_grpc.blockchain_pb2_grpc import BlockchainStub
 
-def main() -> None:
-    # Creating a gRPC channel
-    channel = grpc.insecure_channel("bootstrap1.pactus.org:50051")
 
-    # Creating a stub from channel
+def get_blockchain_info(address: str):
+    channel = grpc.insecure_channel(address)
     stub = BlockchainStub(channel)
-
-    # Initialize a request and call get blockchain info method
     req = GetBlockchainInfoRequest()
     res = stub.GetBlockchainInfo(req)
-
     print(f"Blockchain info:\n{res}")
 
 
 if __name__ == "__main__":
-    main()
+    parser = argparse.ArgumentParser(
+        description="Fetch Pactus blockchain info via gRPC."
+    )
+    parser.add_argument(
+        "--address",
+        default=os.environ.get("GRPC_ADDRESS"),
+        help="gRPC server address (env: GRPC_ADDRESS)",
+    )
+    args = parser.parse_args()
+    if not args.address:
+        parser.error(
+            "No gRPC address provided. Use --address or set GRPC_ADDRESS."
+        )
+
+    get_blockchain_info(args.address)

examples/example_grpc_get_node_info.py

@@ -1,21 +1,42 @@
-from pactus_grpc.network_pb2_grpc import NetworkStub
-from pactus_grpc.network_pb2 import GetNodeInfoRequest
+#!/usr/bin/env python3
+"""
+Pactus SDK Example: Get Node Info via gRPC
+===========================================
+
+This example demonstrates how to connect to a Pactus node via gRPC
+and retrieve node information.
+"""
+
+import argparse
+import os
+
 import grpc
 
+from pactus_grpc.network_pb2 import GetNodeInfoRequest
+from pactus_grpc.network_pb2_grpc import NetworkStub
 
-def main() -> None:
-    # Creating a gRPC channel
-    channel = grpc.insecure_channel("bootstrap1.pactus.org:50051")
 
-    # Creating a stub from channel
+def get_node_info(address: str):
+    channel = grpc.insecure_channel(address)
     stub = NetworkStub(channel)
-
-    # Initialize a request and call get node info method
     req = GetNodeInfoRequest()
     res = stub.GetNodeInfo(req)
-
     print(f"Node info:\n{res}")
 
 
 if __name__ == "__main__":
-    main()
+    parser = argparse.ArgumentParser(
+        description="Fetch Pactus node info via gRPC."
+    )
+    parser.add_argument(
+        "--address",
+        default=os.environ.get("GRPC_ADDRESS"),
+        help="gRPC server address (env: GRPC_ADDRESS)",
+    )
+    args = parser.parse_args()
+    if not args.address:
+        parser.error(
+            "No gRPC address provided. Use --address or set GRPC_ADDRESS."
+        )
+
+    get_node_info(args.address)

examples/example_jsonrpc_get_blockchain_info.py

@@ -1,15 +1,39 @@
-from pactus_jsonrpc.client import PactusOpenRPCClient
+#!/usr/bin/env python3
+"""
+Pactus SDK Example: Get Blockchain Info via JSON-RPC
+=====================================================
+
+This example demonstrates how to connect to a Pactus node via JSON-RPC
+and retrieve blockchain information.
+"""
+
+import argparse
 import asyncio
+import os
 
+from pactus_jsonrpc.client import PactusOpenRPCClient
 
-async def main() -> None:
-    client_url = "https://testnet1.pactus.org/jsonrpc"
-    client = PactusOpenRPCClient(headers={}, client_url=client_url, timeout=300)
 
+async def get_blockchain_info(client_url: str):
+    client = PactusOpenRPCClient(headers={}, client_url=client_url, timeout=300)
     res = await client.pactus.blockchain.get_blockchain_info()
 
     print(f"Blockchain info:\n{res}")
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    parser = argparse.ArgumentParser(
+        description="Fetch Pactus blockchain info via JSON-RPC."
+    )
+    parser.add_argument(
+        "--address",
+        default=os.environ.get("JSONRPC_ADDRESS"),
+        help="JSON-RPC server address (env: JSONRPC_ADDRESS)",
+    )
+    args = parser.parse_args()
+    if not args.address:
+        parser.error(
+            "No JSON-RPC address provided. Use --address or set JSONRPC_ADDRESS."
+        )
+
+    asyncio.run(get_blockchain_info(args.address))

examples/example_jsonrpc_get_node_info.py

@@ -1,15 +1,38 @@
-from pactus_jsonrpc.client import PactusOpenRPCClient
+#!/usr/bin/env python3
+"""
+Pactus SDK Example: Get Node Info via JSON-RPC
+===============================================
+
+This example demonstrates how to connect to a Pactus node via JSON-RPC
+and retrieve node information.
+"""
+
+import argparse
 import asyncio
+import os
+
+from pactus_jsonrpc.client import PactusOpenRPCClient
 
 
-async def main() -> None:
-    client_url = "https://testnet1.pactus.org/jsonrpc"
+async def get_node_info(client_url: str):
     client = PactusOpenRPCClient(headers={}, client_url=client_url, timeout=300)
-
     res = await client.pactus.network.get_node_info()
-
     print(f"Node info:\n{res}")
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    parser = argparse.ArgumentParser(
+        description="Fetch Pactus node info via JSON-RPC."
+    )
+    parser.add_argument(
+        "--address",
+        default=os.environ.get("JSONRPC_ADDRESS"),
+        help="JSON-RPC server address (env: JSONRPC_ADDRESS)",
+    )
+    args = parser.parse_args()
+    if not args.address:
+        parser.error(
+            "No JSON-RPC address provided. Use --address or set JSONRPC_ADDRESS."
+        )
+
+    asyncio.run(get_node_info(args.address))

examples/example_zmq.py

@@ -1,37 +1,57 @@
-import sys
-import zmq
+#!/usr/bin/env python3
+"""
+Pactus SDK Example: Subscribe to ZMQ Topics
+============================================
+
+This example demonstrates how to subscribe to Pactus ZMQ topics.
+"""
+
+import argparse
+import os
 
+import zmq
 
-def main() -> None:
-    if len(sys.argv) < 2:
-        print("Usage: python3 ./example_zmq.py <connect_to> [topic topic ...]")
-        sys.exit(0)
 
-    connect_to = sys.argv[1]
-    topics = sys.argv[2:]
+def subscribe(address: str, topics: list[str]):
     ctx = zmq.Context()
     s = ctx.socket(zmq.SUB)
-    s.connect(connect_to)
+    s.connect(address)
 
-    # manage subscriptions
     if not topics:
         print("Receiving messages on ALL topics...")
         s.setsockopt(zmq.SUBSCRIBE, b"")
     else:
         print(f"Receiving messages on topics: {topics} ...")
         for t in topics:
             s.setsockopt(zmq.SUBSCRIBE, bytes.fromhex(t))
-    print
+    print()
     try:
         while True:
             msg = s.recv_multipart()
             hex_string = [b.hex() for b in msg]
             print("msg: {}".format(hex_string))
-
     except KeyboardInterrupt:
         pass
     print("Done.")
 
 
 if __name__ == "__main__":
-    main()
+    parser = argparse.ArgumentParser(
+        description="Subscribe to Pactus ZMQ topics."
+    )
+    parser.add_argument(
+        "--address",
+        default=os.environ.get("ZMQ_ADDRESS"),
+        help="ZMQ server address (env: ZMQ_ADDRESS)",
+    )
+    parser.add_argument(
+        "topics",
+        nargs="*",
+        help="Optional hex-encoded ZMQ topics to subscribe to",
+    )
+    args = parser.parse_args()
+    if not args.address:
+        print("ZMQ address not set — skipping. Use --address or set ZMQ_ADDRESS.")
+        exit(0)
+
+    subscribe(args.address, args.topics)

pactus/block/block.py

@@ -54,9 +54,9 @@ def id(self) -> Hash:
         return Hash(hashlib.blake2b(buf, digest_size=32).digest())
 
     def _header_bytes(self) -> bytes:
-        writer = Writer()
-        self.header.encode(writer)
-        return writer.bytes()
+        w = Writer()
+        self.header.encode(w)
+        return w.bytes()
 
     def _txs_root(self) -> bytes:
         return Block._merkle_root([tx.id().data for tx in self.transactions])

pactus/block/certificate.py

@@ -58,6 +58,6 @@ def encode(self, writer: Writer) -> None:
 
     def hash(self) -> bytes:
         """Return the certificate hash (blake2b-256 of encoded bytes)."""
-        writer = Writer()
-        self.encode(writer)
-        return hashlib.blake2b(writer.bytes(), digest_size=32).digest()
+        w = Writer()
+        self.encode(w)
+        return hashlib.blake2b(w.bytes(), digest_size=32).digest()

pactus/transaction/transaction.py

@@ -177,26 +177,26 @@ def _get_unsigned_bytes(self, writer: Writer) -> None:
 
     def sign_bytes(self) -> bytes:
         """Return the bytes to be signed (everything except flags)."""
-        writer = Writer()
-        self._get_unsigned_bytes(writer)
-        return writer.bytes()[1:]
+        w = Writer()
+        self._get_unsigned_bytes(w)
+        return w.bytes()[1:]
 
     def id(self) -> Hash:
         """Return the transaction ID (blake2b-256 of sign bytes)."""
         return Hash(hashlib.blake2b(self.sign_bytes(), digest_size=32).digest())
 
     def sign(self, private_key: PrivateKey) -> bytes:
         """Sign the transaction and return signed bytes."""
-        writer = Writer()
-        self._get_unsigned_bytes(writer)
-        sig = private_key.sign(writer.bytes()[1:])
+        w = Writer()
+        self._get_unsigned_bytes(w)
+        sig = private_key.sign(w.bytes()[1:])
         pub = private_key.public_key()
 
-        sig.encode(writer)
-        pub.encode(writer)
+        sig.encode(w)
+        pub.encode(w)
 
         self.public_key = pub
         self.signature = sig
         self.flags |= FLAG_NOT_SIGNED
 
-        return writer.bytes()
+        return w.bytes()