Source code for zpywallet.broadcast.btc.esplora

import requests

from ...errors import NetworkException
from .._provider_success import bitcoin_like_broadcast_success_txid


async def broadcast_transaction_btc_esplora(raw_transaction_hex, **kwargs):
    """Broadcast a Bitcoin transaction using an Esplora node.

    Args:
        raw_transaction_hex (str): The raw transaction in hexadecimal form.
        esplora_url (str): The Esplora endpoint URL.
    """

    esplora_url = kwargs.get("url")
    api_url = f"{esplora_url}/tx"
    headers = {"Content-Type": "text/plain"}

    try:
        response = requests.post(
            api_url, data=raw_transaction_hex, headers=headers, timeout=30
        )
    except Exception as e:
        raise NetworkException(
            "Connection error while broadcasting transaction: {}".format(str(e))
        )

    if response.status_code >= 300:
        raise NetworkException(
            f"Failed to broadcast transaction using Esplora API: {response.text}"
        )

    return bitcoin_like_broadcast_success_txid(raw_transaction_hex)