"""Transaction construction helpers wrapping PyCardano. This module is the SDK's single point of contact with PyCardano's :class:`pycardano.backend.base.ChainContext` API. Everything higher up (``mint`` and eventual refund-path code) goes through the helpers here so we can swap Ogmios for Blockfrost / Cardano-CLI without touching callers. The default context targets the local Ogmios instance on the hot host (``127.0.0.1:1337``). That lines up with the mainnet deployment of the ``cardano-node`` container (v10.6.2 on port 6000 via N2N) fronted by Ogmios as the HTTP+WS bridge. Preprod / testnet callers pass ``network="testnet"`` and typically point at a different host. Cold-signer shape ----------------- ``txbuild`` only knows the hot-side half of the dance: - :func:`make_ogmios_context` — build a context from the live node. - :func:`get_protocol_parameters` — peek at the current protocol params (useful for pricing, ttl calculations, etc.). - :func:`get_address_utxos` — list UTxOs at an address (refund path). - :func:`submit_signed_tx` — ship a tx that was signed offline. Body construction lives in :mod:`cardano_checkout.mint` today. As additional tx shapes (refunds, batched mints) arrive they'll land here alongside ``build_*_tx`` helpers that return :class:`UnsignedMint`-style cold-signer bundles. """ from __future__ import annotations import logging from typing import TYPE_CHECKING, Any, Optional logger = logging.getLogger(__name__) if TYPE_CHECKING: # pragma: no cover — hints only from pycardano import ChainContext, UTxO # --------------------------------------------------------------------------- # Chain context # --------------------------------------------------------------------------- def _require_pycardano() -> None: try: import pycardano # noqa: F401 except ImportError as exc: # pragma: no cover — env sanity raise RuntimeError( "pycardano is required for transaction construction. " "Add pycardano>=0.11.0 to requirements.txt and reinstall." ) from exc def make_ogmios_context( host: str = "127.0.0.1", port: int = 1337, network: str = "mainnet", secure: bool = False, **kwargs: Any, ) -> "ChainContext": """Construct an :class:`pycardano.OgmiosChainContext` for the live node. Args: host: Ogmios HTTP+WS host. Default ``127.0.0.1`` (local). port: Ogmios port. Default ``1337`` (matches the hot host's stack). network: ``"mainnet"`` or ``"testnet"``. Controls the :class:`pycardano.Network` passed to the context. secure: Whether to use wss:// instead of ws://. Default False — the stack assumes a loopback connection. **kwargs: Forwarded to ``OgmiosChainContext`` verbatim (e.g. ``refetch_chain_tip_interval``, ``utxo_cache_size``). Returns: A live :class:`ChainContext`. If the backing node is down the object is still constructed — failures surface on the first query / submit call. """ _require_pycardano() from pycardano import Network, OgmiosChainContext net = Network.MAINNET if network == "mainnet" else Network.TESTNET logger.debug( "[txbuild] OgmiosChainContext -> %s://%s:%d (network=%s)", "wss" if secure else "ws", host, port, network, ) return OgmiosChainContext( host=host, port=port, secure=secure, network=net, **kwargs ) def get_protocol_parameters(context: "ChainContext") -> Any: """Return the live protocol parameters from the chain context. Useful for fee estimation, min-utxo floor computation, and sanity checks that the node is reachable before a mint attempt. The return type is pycardano's :class:`ProtocolParameters` — a dataclass with fields like ``min_fee_a``, ``min_fee_b``, ``coins_per_utxo_byte``, ``max_tx_size``, etc. """ try: return context.protocol_param # type: ignore[attr-defined] except Exception as exc: raise RuntimeError( f"Failed to fetch protocol parameters from chain context: {exc}" ) from exc def get_address_utxos(context: "ChainContext", address: str) -> list["UTxO"]: """Fetch UTxOs at ``address`` via the chain context. Intended for the refund path — when an invoice is cancelled or overpaid the merchant needs to know which UTxOs landed in order to build a return tx. For pure payment-detection, Koios is still the cheaper source (see :mod:`cardano_checkout.monitor`). Args: context: Live chain context (from :func:`make_ogmios_context`). address: Bech32 Cardano address. Returns: List of pycardano :class:`UTxO` objects at ``address``. Empty if the address has no unspent outputs. Never ``None``. Raises: RuntimeError: If the underlying query fails (node down, invalid address). """ _require_pycardano() from pycardano import Address try: addr_obj = Address.from_primitive(address) except Exception as exc: raise RuntimeError(f"Invalid Cardano address: {exc}") from exc try: utxos = context.utxos(str(addr_obj)) # type: ignore[attr-defined] except Exception as exc: raise RuntimeError( f"Failed to fetch UTxOs for {address[:20]}...: {exc}" ) from exc return list(utxos or []) # --------------------------------------------------------------------------- # Signed-tx submission (duplicated from mint.py as a stable txbuild entry # point — the mint module's version delegates here) # --------------------------------------------------------------------------- def submit_signed_tx( signed_tx_cbor_hex: str, context: Optional["ChainContext"] = None, ogmios_host: str = "127.0.0.1", ogmios_port: int = 1337, network: str = "mainnet", ) -> str: """Submit a cold-signed transaction blob to the chain. See :func:`cardano_checkout.mint.submit_signed_tx` for the full docstring — this is the same function under the ``txbuild`` import path so callers that only need submission don't have to import ``mint``. """ _require_pycardano() from pycardano import Transaction if context is None: context = make_ogmios_context( host=ogmios_host, port=ogmios_port, network=network ) try: tx = Transaction.from_cbor(bytes.fromhex(signed_tx_cbor_hex)) except Exception as exc: raise RuntimeError( f"signed_tx_cbor_hex is not valid transaction CBOR: {exc}" ) from exc try: context.submit_tx(tx) # type: ignore[attr-defined] except Exception as exc: raise RuntimeError(f"Ogmios rejected the signed tx: {exc}") from exc tx_hash = str(tx.id) logger.info("[txbuild] submitted signed tx %s", tx_hash) return tx_hash # --------------------------------------------------------------------------- # Placeholders for future tx shapes (kept so consumers can pin imports) # --------------------------------------------------------------------------- def build_payment_tx(*args, **kwargs): # pragma: no cover — future work """Build an unsigned plain-ADA payment tx (refund path). v0.3+.""" raise NotImplementedError( "build_payment_tx lands in v0.3 alongside the refund workflow. " "For v0.2 only mint txs are supported." )