synor/sdk/python/synor_wallet/client.py

"""Synor Wallet Client."""

from typing import Any, Optional
import httpx

from .types import (
    WalletConfig,
    Network,
    WalletType,
    Wallet,
    CreateWalletResult,
    StealthAddress,
    Transaction,
    TransactionInput,
    TransactionOutput,
    SignedTransaction,
    SignedMessage,
    UTXO,
    Balance,
    TokenBalance,
    BalanceResponse,
    FeeEstimate,
    Priority,
    GetUtxosOptions,
    ImportWalletOptions,
    BuildTransactionOptions,
)


class WalletError(Exception):
    """Synor Wallet SDK error."""

    def __init__(
        self,
        message: str,
        status_code: Optional[int] = None,
        code: Optional[str] = None,
    ):
        super().__init__(message)
        self.status_code = status_code
        self.code = code


class SynorWallet:
    """
    Synor Wallet SDK client.

    Example:
        >>> async with SynorWallet(api_key="sk_...") as wallet:
        ...     result = await wallet.create_wallet()
        ...     print(f"Address: {result.wallet.address}")
        ...     print(f"Mnemonic: {result.mnemonic}")  # Store securely!
        ...
        ...     balance = await wallet.get_balance(result.wallet.address)
        ...     print(f"Balance: {balance.native.total}")
    """

    def __init__(
        self,
        api_key: str,
        endpoint: str = "https://wallet.synor.cc/api/v1",
        network: Network = Network.MAINNET,
        timeout: float = 30.0,
        debug: bool = False,
        derivation_path: Optional[str] = None,
    ):
        self.config = WalletConfig(
            api_key=api_key,
            endpoint=endpoint,
            network=network,
            timeout=timeout,
            debug=debug,
            derivation_path=derivation_path,
        )
        self._client = httpx.AsyncClient(
            base_url=endpoint,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
                "X-Network": network.value,
            },
            timeout=timeout,
        )

    async def __aenter__(self) -> "SynorWallet":
        return self

    async def __aexit__(self, *args: Any) -> None:
        await self.close()

    async def close(self) -> None:
        """Close the client."""
        await self._client.aclose()

    async def create_wallet(
        self,
        wallet_type: WalletType = WalletType.STANDARD,
    ) -> CreateWalletResult:
        """
        Create a new wallet.

        Args:
            wallet_type: Type of wallet to create

        Returns:
            Created wallet and mnemonic phrase
        """
        response = await self._request(
            "POST",
            "/wallets",
            {
                "type": wallet_type.value,
                "network": self.config.network.value,
                "derivationPath": self.config.derivation_path,
            },
        )

        wallet = self._parse_wallet(response["wallet"])
        return CreateWalletResult(wallet=wallet, mnemonic=response["mnemonic"])

    async def import_wallet(self, options: ImportWalletOptions) -> Wallet:
        """
        Import a wallet from mnemonic phrase.

        Args:
            options: Import options including mnemonic

        Returns:
            Imported wallet
        """
        response = await self._request(
            "POST",
            "/wallets/import",
            {
                "mnemonic": options.mnemonic,
                "passphrase": options.passphrase,
                "type": options.type.value,
                "network": self.config.network.value,
                "derivationPath": self.config.derivation_path,
            },
        )

        return self._parse_wallet(response["wallet"])

    async def get_wallet(self, wallet_id: str) -> Wallet:
        """
        Get wallet by ID.

        Args:
            wallet_id: Wallet ID

        Returns:
            Wallet details
        """
        response = await self._request("GET", f"/wallets/{wallet_id}")
        return self._parse_wallet(response["wallet"])

    async def list_wallets(self) -> list[Wallet]:
        """
        List all wallets for this account.

        Returns:
            List of wallets
        """
        response = await self._request("GET", "/wallets")
        return [self._parse_wallet(w) for w in response["wallets"]]

    async def get_address(self, wallet_id: str, index: int = 0) -> str:
        """
        Get address at a specific index for a wallet.

        Args:
            wallet_id: Wallet ID
            index: Derivation index

        Returns:
            Address at the index
        """
        response = await self._request(
            "GET", f"/wallets/{wallet_id}/addresses/{index}"
        )
        return response["address"]

    async def get_stealth_address(self, wallet_id: str) -> StealthAddress:
        """
        Generate a stealth address for receiving private payments.

        Args:
            wallet_id: Wallet ID

        Returns:
            Stealth address details
        """
        response = await self._request("POST", f"/wallets/{wallet_id}/stealth")
        sa = response["stealthAddress"]
        return StealthAddress(
            address=sa["address"],
            view_key=sa["viewKey"],
            spend_key=sa["spendKey"],
            ephemeral_key=sa.get("ephemeralKey"),
        )

    async def sign_transaction(
        self,
        wallet_id: str,
        transaction: Transaction,
    ) -> SignedTransaction:
        """
        Sign a transaction.

        Args:
            wallet_id: Wallet ID
            transaction: Transaction to sign

        Returns:
            Signed transaction
        """
        response = await self._request(
            "POST",
            f"/wallets/{wallet_id}/sign",
            {"transaction": self._serialize_transaction(transaction)},
        )

        st = response["signedTransaction"]
        return SignedTransaction(
            raw=st["raw"],
            txid=st["txid"],
            size=st["size"],
            weight=st.get("weight"),
        )

    async def sign_message(
        self,
        wallet_id: str,
        message: str,
        format: str = "text",
    ) -> SignedMessage:
        """
        Sign a message.

        Args:
            wallet_id: Wallet ID
            message: Message to sign
            format: Message format (text, hex, base64)

        Returns:
            Signed message
        """
        response = await self._request(
            "POST",
            f"/wallets/{wallet_id}/sign-message",
            {"message": message, "format": format},
        )

        return SignedMessage(
            signature=response["signature"],
            public_key=response["publicKey"],
            address=response["address"],
        )

    async def verify_message(
        self,
        message: str,
        signature: str,
        address: str,
    ) -> bool:
        """
        Verify a signed message.

        Args:
            message: Original message
            signature: Signature to verify
            address: Expected signer address

        Returns:
            True if signature is valid
        """
        response = await self._request(
            "POST",
            "/verify-message",
            {"message": message, "signature": signature, "address": address},
        )
        return response["valid"]

    async def get_balance(
        self,
        address: str,
        include_tokens: bool = False,
    ) -> BalanceResponse:
        """
        Get balance for an address.

        Args:
            address: Address to check
            include_tokens: Include token balances

        Returns:
            Balance information
        """
        params = {"includeTokens": str(include_tokens).lower()}
        response = await self._request(
            "GET", f"/balances/{address}", params=params
        )

        native = Balance(
            confirmed=response["native"]["confirmed"],
            unconfirmed=response["native"]["unconfirmed"],
            total=response["native"]["total"],
        )

        tokens = []
        if "tokens" in response:
            tokens = [
                TokenBalance(
                    token=t["token"],
                    symbol=t["symbol"],
                    decimals=t["decimals"],
                    balance=t["balance"],
                )
                for t in response["tokens"]
            ]

        return BalanceResponse(native=native, tokens=tokens)

    async def get_utxos(
        self,
        address: str,
        options: Optional[GetUtxosOptions] = None,
    ) -> list[UTXO]:
        """
        Get UTXOs for an address.

        Args:
            address: Address to query
            options: Query options

        Returns:
            List of UTXOs
        """
        params: dict[str, str] = {}
        if options:
            if options.min_confirmations:
                params["minConfirmations"] = str(options.min_confirmations)
            if options.min_amount:
                params["minAmount"] = options.min_amount

        response = await self._request(
            "GET", f"/utxos/{address}", params=params
        )

        return [self._parse_utxo(u) for u in response["utxos"]]

    async def build_transaction(
        self,
        wallet_id: str,
        options: BuildTransactionOptions,
    ) -> Transaction:
        """
        Build a transaction (without signing).

        Args:
            wallet_id: Wallet ID
            options: Transaction building options

        Returns:
            Unsigned transaction
        """
        data: dict[str, Any] = {
            "to": options.to,
            "amount": options.amount,
        }
        if options.fee_rate is not None:
            data["feeRate"] = options.fee_rate
        if options.utxos:
            data["utxos"] = [
                {"txid": u.txid, "vout": u.vout, "amount": u.amount}
                for u in options.utxos
            ]
        if options.change_address:
            data["changeAddress"] = options.change_address

        response = await self._request(
            "POST", f"/wallets/{wallet_id}/build-tx", data
        )

        return self._parse_transaction(response["transaction"])

    async def send_transaction(
        self,
        wallet_id: str,
        options: BuildTransactionOptions,
    ) -> SignedTransaction:
        """
        Build and sign a transaction in one step.

        Args:
            wallet_id: Wallet ID
            options: Transaction building options

        Returns:
            Signed transaction
        """
        tx = await self.build_transaction(wallet_id, options)
        return await self.sign_transaction(wallet_id, tx)

    async def estimate_fee(
        self,
        priority: Priority = Priority.MEDIUM,
    ) -> FeeEstimate:
        """
        Estimate transaction fee.

        Args:
            priority: Priority level

        Returns:
            Fee estimate
        """
        response = await self._request(
            "GET", "/fees/estimate", params={"priority": priority.value}
        )

        return FeeEstimate(
            priority=Priority(response["priority"]),
            fee_rate=response["feeRate"],
            estimated_blocks=response["estimatedBlocks"],
        )

    async def get_all_fee_estimates(self) -> list[FeeEstimate]:
        """
        Get all fee estimates.

        Returns:
            Fee estimates for all priority levels
        """
        response = await self._request("GET", "/fees/estimate/all")

        return [
            FeeEstimate(
                priority=Priority(e["priority"]),
                fee_rate=e["feeRate"],
                estimated_blocks=e["estimatedBlocks"],
            )
            for e in response["estimates"]
        ]

    async def _request(
        self,
        method: str,
        path: str,
        data: Optional[dict[str, Any]] = None,
        params: Optional[dict[str, str]] = None,
    ) -> dict[str, Any]:
        """Make an API request."""
        if self.config.debug:
            print(f"[SynorWallet] {method} {path}")

        response = await self._client.request(
            method,
            path,
            json=data,
            params=params,
        )

        if response.status_code >= 400:
            error = (
                response.json()
                if response.content
                else {"message": response.reason_phrase}
            )
            raise WalletError(
                error.get("message", "Request failed"),
                response.status_code,
                error.get("code"),
            )

        return response.json()

    def _parse_wallet(self, data: dict[str, Any]) -> Wallet:
        """Parse wallet from API response."""
        return Wallet(
            id=data["id"],
            address=data["address"],
            public_key=data["publicKey"],
            type=WalletType(data["type"]),
            created_at=data["createdAt"],
        )

    def _parse_utxo(self, data: dict[str, Any]) -> UTXO:
        """Parse UTXO from API response."""
        return UTXO(
            txid=data["txid"],
            vout=data["vout"],
            amount=data["amount"],
            address=data["address"],
            confirmations=data["confirmations"],
            script_pub_key=data.get("scriptPubKey"),
        )

    def _parse_transaction(self, data: dict[str, Any]) -> Transaction:
        """Parse transaction from API response."""
        inputs = [
            TransactionInput(
                txid=i["txid"],
                vout=i["vout"],
                amount=i["amount"],
                script_sig=i.get("scriptSig"),
            )
            for i in data["inputs"]
        ]
        outputs = [
            TransactionOutput(
                address=o["address"],
                amount=o["amount"],
                script_pub_key=o.get("scriptPubKey"),
            )
            for o in data["outputs"]
        ]
        return Transaction(
            version=data["version"],
            inputs=inputs,
            outputs=outputs,
            lock_time=data.get("lockTime", 0),
            fee=data.get("fee"),
        )

    def _serialize_transaction(self, tx: Transaction) -> dict[str, Any]:
        """Serialize transaction for API request."""
        return {
            "version": tx.version,
            "inputs": [
                {
                    "txid": i.txid,
                    "vout": i.vout,
                    "amount": i.amount,
                }
                for i in tx.inputs
            ],
            "outputs": [
                {
                    "address": o.address,
                    "amount": o.amount,
                }
                for o in tx.outputs
            ],
            "lockTime": tx.lock_time,
        }