"""TransactionBuilder for constructing, signing, serializing, and broadcasting BCH transactions."""
import asyncio
from typing import cast
from cashscript_py.helpers.transaction import create_op_return_output, serialize_transaction, validate_output
from cashscript_py.interfaces import Output, Transaction, UnlockableUtxo, Unlocker, Utxo
from cashscript_py.network.network_provider import NetworkProvider
[docs]
class TransactionBuilder:
"""Build, sign, and send BCH transactions via unlockers.
Supports multiple inputs/outputs, OP_RETURN, fee caps, and token burn checks.
"""
def __init__(
self,
provider: NetworkProvider,
*,
maximum_fee_satoshis: int | None = None,
maximum_fee_sats_per_byte: float | None = None,
allow_implicit_fungible_token_burn: bool = False,
):
"""Initialize a TransactionBuilder.
Args:
provider: NetworkProvider used for queries and broadcasts.
maximum_fee_satoshis: Absolute fee cap (satoshis).
maximum_fee_sats_per_byte: Per-byte fee cap (sats/byte).
allow_implicit_fungible_token_burn: Allow outputs to burn fungible tokens.
"""
self.provider: NetworkProvider = provider
self.maximum_fee_satoshis: int | None = maximum_fee_satoshis
self.maximum_fee_sats_per_byte: float | None = maximum_fee_sats_per_byte
self.allow_implicit_fungible_token_burn: bool = allow_implicit_fungible_token_burn
self.inputs: list[UnlockableUtxo] = []
self.outputs: list[Output] = []
self.locktime: int = 0
self.version: int = 2
[docs]
def add_output(self, output: Output) -> "TransactionBuilder":
"""Add a single output.
Args:
output: Recipient and amount, optionally with token details.
Returns:
This builder (for chaining).
"""
return self.add_outputs([output])
[docs]
def add_outputs(self, outputs: list[Output]) -> "TransactionBuilder":
"""Add multiple outputs (validated for dust, tokens, etc.).
Args:
outputs: Outputs to append.
Returns:
This builder (for chaining).
Raises:
OutputAddressInvalid: If a provided address cannot be parsed.
OutputSatoshisNonPositiveError: If amount <= 0.
OutputSatoshisTooSmallError: If amount is below dust minimum.
TokensToNonTokenAddressError: If sending tokens to a non-token address.
OutputTokenAmountTooSmallError: If token amount is negative.
"""
for o in outputs:
validate_output(o)
self.outputs.extend(outputs)
return self
[docs]
def add_op_return_output(self, chunks: list[str]) -> "TransactionBuilder":
"""Add an OP_RETURN output.
Args:
chunks: List of strings (UTF-8 or hex with 0x prefix) to push.
"""
self.outputs.append(create_op_return_output(chunks))
return self
[docs]
def set_locktime(self, locktime: int) -> "TransactionBuilder":
"""Set the transaction locktime.
Args:
locktime: Block height or timestamp (per network rules).
Returns:
This builder (for chaining).
"""
self.locktime = locktime
return self
def _check_fungible_token_burn(self) -> None:
"""Disallow implicit burning of fungible tokens unless explicitly allowed.
For each category, total output amount must be >= total input amount.
"""
if self.allow_implicit_fungible_token_burn:
return
token_input_amounts: dict[str, int] = {}
token_output_amounts: dict[str, int] = {}
for i in self.inputs:
t = i.token
if t and t.amount and t.amount > 0:
token_input_amounts[t.category] = token_input_amounts.get(t.category, 0) + t.amount
for o in self.outputs:
t = o.token
if t and t.amount and t.amount > 0:
token_output_amounts[t.category] = token_output_amounts.get(t.category, 0) + t.amount
for category, input_amount in token_input_amounts.items():
output_amount = token_output_amounts.get(category, 0)
if output_amount < input_amount:
raise ValueError(
f"Implicit burning of fungible tokens for category {category} is not allowed "
f"(input amount: {input_amount}, output amount: {output_amount}). "
f"If this is intended, set allow_implicit_fungible_token_burn=True."
)
def _check_max_fee(self, encoded_tx_bytes: bytes) -> None:
"""Enforce maximum_fee_satoshis and maximum_fee_sats_per_byte, if set."""
total_input_amount = sum(i.satoshis for i in self.inputs)
total_output_amount = sum(o.amount for o in self.outputs)
fee = total_input_amount - total_output_amount
if self.maximum_fee_satoshis is not None and fee > self.maximum_fee_satoshis:
raise ValueError(f"Transaction fee of {fee} is higher than max fee of {self.maximum_fee_satoshis}")
if self.maximum_fee_sats_per_byte is not None:
tx_size = len(encoded_tx_bytes)
fee_per_byte = float(f"{(fee / tx_size):.2f}") if tx_size > 0 else float("inf")
if fee_per_byte > self.maximum_fee_sats_per_byte:
raise ValueError(
f"Transaction fee per byte of {fee_per_byte} is higher than max fee per byte of "
f"{self.maximum_fee_sats_per_byte}"
)
[docs]
def build(self) -> str:
"""Build the transaction, generate unlocking scripts, and return serialized hex.
Returns:
Hex-encoded transaction.
Raises:
ValueError: If fee caps are exceeded or implicit token burns are disallowed.
"""
# Disallow unintended token burns
self._check_fungible_token_burn()
# Create the transaction template for signing
tx = Transaction(self.inputs, self.locktime, self.outputs, self.version)
# Generate source outputs (needed for SIGHASH_UTXOS)
source_outputs: list[Output] = []
for utxo in self.inputs:
source = Output(utxo.unlocker.generate_locking_bytecode(), utxo.satoshis, utxo.token)
source_outputs.append(source)
# Generate the final unlocking scripts for each input
for i, input in enumerate(self.inputs):
input.unlocking_bytecode = input.unlocker.generate_unlocking_bytecode(tx, source_outputs, i)
# Serialize, then check fee constraints with final size
encoded_tx_bytes = serialize_transaction(tx)
self._check_max_fee(encoded_tx_bytes)
# Return hex
return encoded_tx_bytes.hex()
[docs]
async def send(self, raw: bool = False) -> dict[str, str]:
"""Build and broadcast the transaction, retrying for details.
Args:
raw: If True, return only the raw hex when details are fetched.
Returns:
A dict containing either {'hex': ...} when raw=True or {'txid': ..., 'hex': ...}.
Raises:
Exception: If broadcasting fails.
"""
tx_hex = self.build()
txid = await self.provider.send_raw_transaction(tx_hex)
return await self.get_tx_details(txid, raw)
[docs]
async def get_tx_details(self, txid: str, raw: bool = False) -> dict[str, str]:
"""Poll for transaction details until available or timeout.
Args:
txid: Transaction ID to fetch.
raw: If True, return only hex (no decoded details).
Returns:
A dict containing either {'hex': ...} when raw=True or {'txid': ..., 'hex': ...}.
Raises:
Exception: If details cannot be retrieved in time.
"""
retries = 0
max_retries = 1200
delay_time = 0.5
while retries < max_retries:
await asyncio.sleep(delay_time)
try:
tx_hex = await self.provider.get_raw_transaction(txid)
if raw:
return {"hex": tx_hex}
return {
"txid": txid,
"hex": tx_hex,
}
except Exception:
retries += 1
raise Exception("Transaction details could not be retrieved after multiple attempts.")