"""Network provider interfaces and enums."""
from abc import ABC, abstractmethod
from enum import StrEnum
from typing import Any
from cashscript_py.interfaces import Utxo
[docs]
class Network(StrEnum):
"""Supported Bitcoin Cash networks (network and address prefixes)."""
MAINNET = "mainnet"
TESTNET3 = "testnet3"
TESTNET4 = "testnet4"
CHIPNET = "chipnet"
MOCKNET = "mocknet"
REGTEST = "regtest"
[docs]
class BroadcastFailed(RuntimeError):
"""Raised when a network broadcast attempt fails."""
def __init__(self, error: Exception | str) -> None:
"""Initialize with the broadcast error detail."""
super().__init__(f"Broadcast failed: `{error}`")
[docs]
class RequestFailed(RuntimeError):
"""Raised when a network request fails."""
def __init__(self, error: Exception):
"""Initialize with the request error detail."""
super().__init__(f"Request failed: `{error}`")
[docs]
class UnexpectedResponse(RuntimeError):
"""Raised when a network response has an unexpected format or type."""
def __init__(self, response: Any):
"""Initialize with the unexpected response value."""
super().__init__(f"Unexpected reponse from server: `{response}`")
[docs]
class NetworkProvider(ABC):
"""Abstract base class for providers connecting to BCH networks.
Implementations must support querying UTXOs, block height, raw transactions, and broadcasting.
"""
@property
@abstractmethod
def network(self) -> Network:
"""Return the network this provider connects to."""
pass # pragma: no cover
[docs]
@abstractmethod
async def get_utxos(self, address: str) -> list[Utxo]:
"""Retrieve UTXOs for a given CashAddress.
Args:
address: The CashAddress for which UTXOs are requested.
Returns:
A list of spendable UTXOs.
"""
pass # pragma: no cover
[docs]
@abstractmethod
async def get_block_height(self) -> int:
"""Return the current block height.
Raises:
RequestFailed: If the provider returns an error.
UnexpectedResponse: If the provider returns a malformed response.
"""
pass # pragma: no cover
[docs]
@abstractmethod
async def get_raw_transaction(self, txid: str) -> str:
"""Retrieve the raw transaction hex for a txid.
Args:
txid: Transaction ID (hex).
Returns:
Raw transaction hex.
Raises:
RequestFailed: If the provider returns an error.
UnexpectedResponse: If the provider returns a malformed response.
"""
pass # pragma: no cover
[docs]
@abstractmethod
async def send_raw_transaction(self, tx_hex: str) -> str:
"""Broadcast a raw transaction to the network.
Args:
tx_hex: Hex-encoded transaction.
Returns:
Transaction ID computed for the broadcast transaction.
Raises:
BroadcastFailed: If the transaction was not accepted by the network.
"""
pass # pragma: no cover