prime-sdk-python/prime_sdk/models.py

"""Request and response models, ported from the Go SDK's ``models/models.go``.

Request models expose ``to_dict()`` which emits the exact JSON keys the node
expects and drops empty/optional values (the Go ``omitempty`` equivalent).
Response models expose a ``from_dict()`` classmethod that reads the exact JSON
keys returned by the node (snake_case for transaction/block models, camelCase
for contract/status models — preserved verbatim from the Go struct tags).
"""

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

CONTENT_TYPE_JSON = "application/json"


# --------------------------------------------------------------------------- #
# Transactions
# --------------------------------------------------------------------------- #


@dataclass
class TransactionCreateRequest:
    txn_type: str
    payload: str
    tag: Optional[str] = None
    version: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        d: Dict[str, Any] = {"txn_type": self.txn_type, "payload": self.payload}
        if self.version:
            d["version"] = self.version
        if self.tag:
            d["tag"] = self.tag
        return d


@dataclass
class TransactionCreateResponse:
    transaction_id: str = ""

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "TransactionCreateResponse":
        return cls(transaction_id=d.get("transaction_id", ""))


@dataclass
class TransactionBulkRequest:
    transactions: List[TransactionCreateRequest] = field(default_factory=list)

    def to_dict(self) -> Dict[str, Any]:
        return {"transactions": [t.to_dict() for t in self.transactions]}


@dataclass
class TransactionBulkResponse:
    transaction_ids: List[str] = field(default_factory=list)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "TransactionBulkResponse":
        return cls(transaction_ids=d.get("transaction_ids") or [])


@dataclass
class TransactionHeader:
    tag: str = ""
    dc_id: str = ""
    txn_id: str = ""
    invoker: str = ""
    block_id: str = ""
    txn_type: str = ""
    timestamp: str = ""

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "TransactionHeader":
        return cls(
            tag=d.get("tag", ""),
            dc_id=d.get("dc_id", ""),
            txn_id=d.get("txn_id", ""),
            invoker=d.get("invoker", ""),
            block_id=d.get("block_id", ""),
            txn_type=d.get("txn_type", ""),
            timestamp=d.get("timestamp", ""),
        )


@dataclass
class TransactionProof:
    full: str = ""
    stripped: str = ""

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "TransactionProof":
        return cls(full=d.get("full", ""), stripped=d.get("stripped", ""))


@dataclass
class Transaction:
    version: str = ""
    header: TransactionHeader = field(default_factory=TransactionHeader)
    proof: TransactionProof = field(default_factory=TransactionProof)
    payload: str = ""

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "Transaction":
        return cls(
            version=d.get("version", ""),
            header=TransactionHeader.from_dict(d.get("header") or {}),
            proof=TransactionProof.from_dict(d.get("proof") or {}),
            payload=d.get("payload", ""),
        )


@dataclass
class ListTransactionsResponse:
    transactions: List[Transaction] = field(default_factory=list)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "ListTransactionsResponse":
        return cls(
            transactions=[Transaction.from_dict(t) for t in (d.get("transactions") or [])]
        )


# --------------------------------------------------------------------------- #
# Transaction types
# --------------------------------------------------------------------------- #


@dataclass
class TransactionTypeCreateRequest:
    txn_type: str
    version: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        d: Dict[str, Any] = {"txn_type": self.txn_type}
        if self.version:
            d["version"] = self.version
        return d


@dataclass
class TransactionTypeCreateResponse:
    success: bool = False

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "TransactionTypeCreateResponse":
        return cls(success=bool(d.get("success", False)))


@dataclass
class TransactionType:
    version: str = ""
    created: int = 0
    modified: int = 0
    txn_type: str = ""
    contract_id: str = ""
    custom_indexes: List[Any] = field(default_factory=list)
    active_since_block: str = ""

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "TransactionType":
        return cls(
            version=d.get("version", ""),
            created=d.get("created", 0),
            modified=d.get("modified", 0),
            txn_type=d.get("txn_type", ""),
            contract_id=d.get("contract_id", ""),
            custom_indexes=d.get("custom_indexes") or [],
            active_since_block=d.get("active_since_block", ""),
        )


@dataclass
class TransactionListResponse:
    """Response for listing transaction types (key: ``transactionTypes``)."""

    transaction_types: List[TransactionType] = field(default_factory=list)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "TransactionListResponse":
        return cls(
            transaction_types=[
                TransactionType.from_dict(t) for t in (d.get("transactionTypes") or [])
            ]
        )


# --------------------------------------------------------------------------- #
# Smart contracts
# --------------------------------------------------------------------------- #


@dataclass
class SmartContractCreateRequest:
    environment: str
    transaction_type: str
    execution_order: str
    environment_variables: Optional[Dict[str, str]] = None
    secrets: Optional[Dict[str, str]] = None
    remote: bool = False

    def to_dict(self) -> Dict[str, Any]:
        d: Dict[str, Any] = {
            "environment": self.environment,
            "transactionType": self.transaction_type,
            "executionOrder": self.execution_order,
        }
        if self.environment_variables:
            d["environmentVariables"] = self.environment_variables
        if self.secrets:
            d["secret"] = self.secrets
        if self.remote:
            d["remote"] = self.remote
        return d


@dataclass
class SmartContractUpdateRequest:
    enabled: bool = False
    environment_variables: Optional[Dict[str, str]] = None
    secrets: Optional[Dict[str, str]] = None
    version: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        d: Dict[str, Any] = {"enabled": self.enabled}
        if self.version:
            d["version"] = self.version
        if self.environment_variables:
            d["environmentVariables"] = self.environment_variables
        if self.secrets:
            d["secret"] = self.secrets
        return d


@dataclass
class GrpcConnectionInfo:
    address: str = ""
    api_key: str = ""

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "GrpcConnectionInfo":
        return cls(address=d.get("address", ""), api_key=d.get("apiKey", ""))


@dataclass
class SmartContractExecutionInfo:
    type: str = ""
    executable_path: str = ""
    executable_working_directory: str = ""
    executable_hash: str = ""

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "SmartContractExecutionInfo":
        return cls(
            type=d.get("type", ""),
            executable_path=d.get("executablePath", ""),
            executable_working_directory=d.get("executableWorkingDirectory", ""),
            executable_hash=d.get("executableHash", ""),
        )


@dataclass
class SmartContract:
    id: str = ""
    created: int = 0
    modified: int = 0
    version: str = ""
    environment: str = ""
    transaction_type: str = ""
    execution_order: str = ""
    execution_info: Optional[SmartContractExecutionInfo] = None
    env_vars: Dict[str, str] = field(default_factory=dict)
    secrets: List[str] = field(default_factory=list)
    grpc_connection_info: Optional[GrpcConnectionInfo] = None

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "SmartContract":
        exec_info = d.get("executionInfo")
        grpc = d.get("grpcConnectionInfo")
        return cls(
            id=d.get("id", ""),
            created=d.get("created", 0),
            modified=d.get("modified", 0),
            version=d.get("version", ""),
            environment=d.get("environment", ""),
            transaction_type=d.get("transactionType", ""),
            execution_order=d.get("executionOrder", ""),
            execution_info=(
                SmartContractExecutionInfo.from_dict(exec_info) if exec_info else None
            ),
            env_vars=d.get("envVars") or {},
            secrets=d.get("secrets") or [],
            grpc_connection_info=(GrpcConnectionInfo.from_dict(grpc) if grpc else None),
        )


# --------------------------------------------------------------------------- #
# Blocks
# --------------------------------------------------------------------------- #


@dataclass
class BlockProof:
    proof: str = ""
    scheme: str = ""  # absent on trust-scheme chains; present on PoW
    nonce: int = 0

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "BlockProof":
        return cls(
            proof=d.get("proof", ""),
            scheme=d.get("scheme", ""),
            nonce=d.get("nonce", 0),
        )


@dataclass
class BlockHeader:
    block_id: str = ""
    dc_id: str = ""
    prev_id: str = ""
    prev_proof: str = ""
    timestamp: str = ""

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "BlockHeader":
        return cls(
            block_id=d.get("blockId", ""),
            dc_id=d.get("dcId", ""),
            prev_id=d.get("prevId", ""),
            prev_proof=d.get("prevProof", ""),
            timestamp=d.get("timestamp", ""),
        )


@dataclass
class Block:
    version: str = ""
    header: BlockHeader = field(default_factory=BlockHeader)
    transactions: List[str] = field(default_factory=list)
    proof: BlockProof = field(default_factory=BlockProof)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "Block":
        return cls(
            version=d.get("version", ""),
            header=BlockHeader.from_dict(d.get("header") or {}),
            transactions=d.get("transactions") or [],
            proof=BlockProof.from_dict(d.get("proof") or {}),
        )


# --------------------------------------------------------------------------- #
# System / generic
# --------------------------------------------------------------------------- #


@dataclass
class SystemStatus:
    id: str = ""
    level: int = 0
    url: str = ""
    hash_algo: str = ""
    scheme: str = ""
    version: str = ""
    encryption_algo: str = ""
    indexing_enabled: bool = False
    # Level 5 node fields
    funded: str = ""
    broadcast_interval: str = ""
    network: str = ""
    interchain_wallet: str = ""

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "SystemStatus":
        return cls(
            id=d.get("id", ""),
            level=d.get("level", 0),
            url=d.get("url", ""),
            hash_algo=d.get("hashAlgo", ""),
            scheme=d.get("scheme", ""),
            version=d.get("version", ""),
            encryption_algo=d.get("encryptionAlgo", ""),
            indexing_enabled=bool(d.get("indexingEnabled", False)),
            funded=d.get("funded", ""),
            broadcast_interval=d.get("broadcastInterval", ""),
            network=d.get("network", ""),
            interchain_wallet=d.get("interchainWallet", ""),
        )


@dataclass
class SuccessResponse:
    success: bool = False

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "SuccessResponse":
        return cls(success=bool(d.get("success", False)))


@dataclass
class ListResponse:
    items: List[Any] = field(default_factory=list)
    total_count: int = 0

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "ListResponse":
        return cls(items=d.get("items") or [], total_count=d.get("total_count", 0))