refactor conftest.py to use ProviderBuilder

Author: Matthias Zimmermann

README.md

@@ -27,12 +27,13 @@ Here's a "Hello World!" example showing how to use the Python Arkiv SDK:
 from web3 import HTTPProvider
 from arkiv import Arkiv
 from arkiv.account import NamedAccount
+from arkiv.provider import ProviderBuilder
 
 with open ('wallet_alice.json', 'r') as f:
     wallet = f.read()
 
 # Initialize Arkiv client (extends Web3)
-provider = HTTPProvider('https://kaolin.hoodi.arkiv.network/rpc')
+provider = ProviderBuilder().kaolin().build()
 account = NamedAccount.from_wallet('Alice', wallet, 's3cret')
 client = Arkiv(provider, account = account)
 

src/arkiv/provider.py

@@ -1,6 +1,6 @@
 """Provider builder for creating Web3 providers with Arkiv presets."""
 
-from typing import Literal
+from typing import Literal, cast
 
 from web3.providers import HTTPProvider, WebSocketProvider
 from web3.providers.base import BaseProvider
@@ -45,8 +45,8 @@ class ProviderBuilder:
 
     def __init__(self) -> None:
         """Initialize the provider builder."""
-        self._network: str = NETWORK_DEFAULT
-        self._transport: TransportType = TRANSPORT_DEFAULT
+        self._network: str | None = NETWORK_DEFAULT
+        self._transport: TransportType = cast(TransportType, TRANSPORT_DEFAULT)
         self._port: int | None = DEFAULT_PORT  # Set default port for localhost
         self._url: str | None = None
 
@@ -99,7 +99,7 @@ def http(self) -> "ProviderBuilder":
         Returns:
             Self for method chaining
         """
-        self._transport = HTTP
+        self._transport = cast(TransportType, HTTP)
         return self
 
     def ws(self) -> "ProviderBuilder":
@@ -109,7 +109,7 @@ def ws(self) -> "ProviderBuilder":
         Returns:
             Self for method chaining
         """
-        self._transport = WS
+        self._transport = cast(TransportType, WS)
         return self
 
     def build(self) -> BaseProvider:
@@ -123,6 +123,7 @@ def build(self) -> BaseProvider:
         Raises:
             ValueError: If no URL has been configured or if transport is not available
         """
+        url: str
         # Custom URL overrides network constant
         if self._url is not None:
             url = self._url
@@ -132,13 +133,14 @@ def build(self) -> BaseProvider:
             if network_urls is None:
                 raise ValueError(f"Unknown network: {self._network}")
 
-            url = network_urls.get(self._transport)
-            if url is None:
+            url_from_network = network_urls.get(self._transport)
+            if url_from_network is None:
                 available = ", ".join(network_urls.keys())
                 raise ValueError(
                     f"Transport '{self._transport}' is not available for network '{self._network}'. "
                     f"Available transports: {available}"
                 )
+            url = url_from_network
 
             if self._port is not None:
                 # Append port only for localhost
@@ -154,4 +156,4 @@ def build(self) -> BaseProvider:
         if self._transport == HTTP:
             return HTTPProvider(url)
         else:
-            return WebSocketProvider(url)
+            return cast(BaseProvider, WebSocketProvider(url))

tests/conftest.py

@@ -12,9 +12,11 @@
 import pytest
 from testcontainers.core.container import DockerContainer
 from web3 import Web3
+from web3.providers.base import BaseProvider
 
 from arkiv import Arkiv
 from arkiv.account import NamedAccount
+from arkiv.provider import ProviderBuilder
 from tests.node_container import create_container_node
 
 from .utils import create_account
@@ -46,6 +48,30 @@ def _load_env_if_available() -> None:
         logger.debug("python-dotenv not available, using system environment only")
 
 
+def check_and_get_web3(name: str, provider: BaseProvider) -> Web3:
+    """Create and verify a Web3 client connection."""
+    return verify_connection(name, Web3(provider))
+
+
+def check_and_get_arkiv(
+    name: str, provider: BaseProvider, account: NamedAccount | None = None
+) -> Arkiv:
+    """Create and verify an Arkiv client connection."""
+    if account:
+        return verify_connection(name, Arkiv(provider, account=account))
+    else:
+        return verify_connection(name, Arkiv(provider))
+
+
+def verify_connection(name: str, client: Web3) -> Web3:
+    """Verify Web3 client connection and return checked client."""
+    assert client.is_connected(), (
+        f"{name}: Failed to connect to {client.provider.endpoint_uri}"
+    )  # type: ignore[attr-defined]
+    logger.info(f"{name} client connected to {client.provider.endpoint_uri}")  # type: ignore[attr-defined]
+    return client
+
+
 # Load environment on import
 _load_env_if_available()
 
@@ -74,59 +100,44 @@ def arkiv_node() -> Generator[tuple[DockerContainer | None, str, str], None, Non
 
 
 @pytest.fixture(scope="session")
-def web3_client_http(arkiv_node: tuple[DockerContainer | None, str, str]) -> Web3:
-    """
-    Provide Web3 client connected to HTTP endpoint.
-
-    Returns:
-        Web3 client instance connected to the arkiv node
-    """
+def testcontainer_provider_http(
+    arkiv_node: tuple[DockerContainer | None, str, str],
+) -> BaseProvider:
+    """Returns testcontainer HTTPProvider using the ProviderBuilder."""
     _, http_url, _ = arkiv_node
-    client = Web3(Web3.HTTPProvider(http_url))
+    return ProviderBuilder().custom(http_url).build()
 
-    # Verify connection
-    assert client.is_connected(), f"Failed to connect to {http_url}"
 
-    logger.info(f"Web3 client connected to {http_url}")
-    return client
+@pytest.fixture(scope="session")
+def testcontainer_provider_ws(
+    arkiv_node: tuple[DockerContainer | None, str, str],
+) -> BaseProvider:
+    """Returns testcontainer WebSocketProvider using the ProviderBuilder."""
+    _, _, ws_url = arkiv_node
+    # Note: .ws() is to make transport explicit, it is not strictly necessary as the URL scheme is ws:// or wss://
+    return ProviderBuilder().custom(ws_url).ws().build()
 
 
 @pytest.fixture(scope="session")
-def arkiv_ro_client_http(arkiv_node: tuple[DockerContainer | None, str, str]) -> Arkiv:
-    """
-    Provide Arkiv client connected to HTTP endpoint.
+def web3_client_http(testcontainer_provider_http: BaseProvider) -> Web3:
+    """Return Web3 client connected to HTTP endpoint."""
+    return check_and_get_web3("Web3 HTTP", testcontainer_provider_http)
 
-    Returns:
-        Web3 client instance connected to the arkiv node
-    """
-    _, http_url, _ = arkiv_node
-    client = Arkiv(Web3.HTTPProvider(http_url))
 
-    # Verify connection
-    assert client.is_connected(), f"Failed to connect Arkiv client to {http_url}"
-
-    logger.info(f"Arkiv client connected to {http_url}")
-    return client
+@pytest.fixture(scope="session")
+def arkiv_ro_client_http(testcontainer_provider_http: BaseProvider) -> Arkiv:
+    """Return a read only Arkiv client connected to HTTP endpoint."""
+    return check_and_get_arkiv("Arkiv HTTP (read only)", testcontainer_provider_http)
 
 
 @pytest.fixture(scope="session")
 def arkiv_client_http(
-    arkiv_node: tuple[DockerContainer | None, str, str], account_1: NamedAccount
+    testcontainer_provider_http: BaseProvider, account_1: NamedAccount
 ) -> Arkiv:
-    """
-    Provide Arkiv client connected to HTTP endpoint.
-
-    Returns:
-        Web3 client instance connected to the arkiv node
-    """
-    _, http_url, _ = arkiv_node
-    client = Arkiv(Web3.HTTPProvider(http_url), account=account_1)
-
-    # Verify connection
-    assert client.is_connected(), f"Failed to connect Arkiv client to {http_url}"
-
-    logger.info(f"Arkiv client connected to {http_url}")
-    return client
+    """Return a read only Arkiv client connected to HTTP endpoint."""
+    return check_and_get_arkiv(
+        "Arkiv HTTP", testcontainer_provider_http, account=account_1
+    )
 
 
 @pytest.fixture(scope="session")