Andrew Miller
e4e49218d0
New transaction.get_interchain and block.get_interchain call the prime-node
/api/v1/{transaction,block}/{id}/interchain endpoints, returning an
InterchainTrace {block_id, validator_blocks, interchain_transactions}. Adds
VerificationBlock / InterchainTransaction / InterchainTrace dataclasses with
from_dict, exports them, and a from_dict test.
495 lines
15 KiB
Python
495 lines
15 KiB
Python
"""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 {}),
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Interchain trace
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@dataclass
|
|
class VerificationBlock:
|
|
"""A validator's verification of a prime block."""
|
|
|
|
version: str = ""
|
|
prime_chain_id: str = ""
|
|
prime_block_id: str = ""
|
|
timestamp: str = ""
|
|
verifier_public_key: str = ""
|
|
verifier_signature: str = ""
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: Dict[str, Any]) -> "VerificationBlock":
|
|
return cls(
|
|
version=d.get("version", ""),
|
|
prime_chain_id=d.get("primeChainId", ""),
|
|
prime_block_id=d.get("primeBlockId", ""),
|
|
timestamp=d.get("timestamp", ""),
|
|
verifier_public_key=d.get("verifierPublicKey", ""),
|
|
verifier_signature=d.get("verifierSignature", ""),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class InterchainTransaction:
|
|
"""An anchor broadcast to a public blockchain (e.g. ETH/BTC) bundling one or
|
|
more validator blocks. ``validator_blocks`` holds the covered prime block ids;
|
|
``covered_prime_chain_ids`` the prime chains they belong to."""
|
|
|
|
id: int = 0
|
|
version: str = ""
|
|
timestamp: str = ""
|
|
chain_id: str = ""
|
|
trans_hash: str = ""
|
|
block_id: str = ""
|
|
validator_blocks: List[str] = field(default_factory=list)
|
|
validator_blockhash: str = ""
|
|
signature: str = ""
|
|
covered_prime_chain_ids: List[str] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: Dict[str, Any]) -> "InterchainTransaction":
|
|
return cls(
|
|
id=d.get("id", 0),
|
|
version=d.get("version", ""),
|
|
timestamp=d.get("timestamp", ""),
|
|
chain_id=d.get("chainId", ""),
|
|
trans_hash=d.get("transHash", ""),
|
|
block_id=d.get("blockId", ""),
|
|
validator_blocks=d.get("validatorBlocks") or [],
|
|
validator_blockhash=d.get("validatorBlockhash", ""),
|
|
signature=d.get("signature", ""),
|
|
covered_prime_chain_ids=d.get("coveredPrimeChainIds") or [],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class InterchainTrace:
|
|
"""Links a prime block to the validator (verification) blocks that validated
|
|
it and the public-chain interchain anchors those validator blocks were
|
|
bundled into. Returned by ``transaction.get_interchain`` and
|
|
``block.get_interchain``."""
|
|
|
|
block_id: str = ""
|
|
validator_blocks: List[VerificationBlock] = field(default_factory=list)
|
|
interchain_transactions: List[InterchainTransaction] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: Dict[str, Any]) -> "InterchainTrace":
|
|
return cls(
|
|
block_id=d.get("blockId", ""),
|
|
validator_blocks=[
|
|
VerificationBlock.from_dict(v) for v in (d.get("validatorBlocks") or [])
|
|
],
|
|
interchain_transactions=[
|
|
InterchainTransaction.from_dict(v)
|
|
for v in (d.get("interchainTransactions") 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))
|