Files
prime-sdk-python/prime_sdk/models.py
Andrew Miller 7b1e9f6309
All checks were successful
Publish to PyPI Registry / publish (release) Successful in 1m14s
Add proof-measure client + Gitea PyPI publish; bump to 0.4.0
proof-measure is a separate, public, unauthenticated Dragonchain service. Adds:
- UnauthenticatedClient: HMAC-free transport mirroring Client (session
  injection, allow_redirects=False, from_dict decoding).
- ProofMeasureClient: get_security / report / health; default base URL
  https://proof-measure.dragonchain.com. Standalone (ProofMeasureClient()) and
  via DragonchainSDK.proof_measure.
- Proof-measure dataclass models (decimals as strings, timestamps as int).
- .gitea/workflows/publish.yml: build + twine upload to the Gitea PyPI registry
  on release (the SDK had no publish workflow before).

Also fixes 3 pre-existing failing tests: _FakeSession.request didn't accept the
allow_redirects kwarg the client now passes (added by the prior redirect change),
so the suite was red at HEAD.
2026-06-04 13:53:16 -04:00

661 lines
20 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))
# --------------------------------------------------------------------------- #
# Proof Measure (measured immutability / "securedBy")
# --------------------------------------------------------------------------- #
#
# Decimal-valued fields are kept as strings (full precision); timestamps are int
# unix seconds. JSON keys are camelCase, preserved verbatim from the service.
@dataclass
class RawMeasure:
"""A network's native cumulative security measure: cumulative hashes for
PoW, stake-seconds for PoS."""
value: str = "" # human-scaled mantissa, e.g. "484.44"
unit: str = "" # e.g. "Zettahashes", "ETH·s"
base: str = "" # unscaled base amount (hashes or stake-seconds)
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "RawMeasure":
return cls(
value=d.get("value", ""),
unit=d.get("unit", ""),
base=d.get("base", ""),
)
@dataclass
class SecurityResult:
"""Security a single network accrued for a window/anchor, as the raw native
measure AND a USD valuation, plus a normalized 0..1 score."""
network: str = ""
consensus: str = "" # "pow" | "pos"
raw: RawMeasure = field(default_factory=RawMeasure)
value_usd: str = ""
value_usd_formatted: str = ""
label: str = ""
normalized_score: str = ""
since: int = 0 # window/anchor start (unix seconds)
as_of: int = 0 # latest sample time (unix seconds)
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "SecurityResult":
return cls(
network=d.get("network", ""),
consensus=d.get("consensus", ""),
raw=RawMeasure.from_dict(d.get("raw") or {}),
value_usd=d.get("valueUsd", ""),
value_usd_formatted=d.get("valueUsdFormatted", ""),
label=d.get("label", ""),
normalized_score=d.get("normalizedScore", ""),
since=d.get("since", 0),
as_of=d.get("asOf", 0),
)
@dataclass
class ReportAnchorInput:
"""One anchor supplied in a report request. Provide either ``network``
("BTC"/"ETH") or the public-chain numeric ``chain_id``; network wins."""
tx_hash: str
timestamp: int # anchor time, unix seconds
network: Optional[str] = None
chain_id: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {"txHash": self.tx_hash, "timestamp": self.timestamp}
if self.network:
d["network"] = self.network
if self.chain_id:
d["chainId"] = self.chain_id
return d
@dataclass
class ReportRequest:
"""Body of ``ProofMeasureClient.report``: a transaction's interchain anchors."""
transaction_id: str
prime_id: str
block_id: str
anchors: List[ReportAnchorInput] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
return {
"transactionId": self.transaction_id,
"primeId": self.prime_id,
"blockId": self.block_id,
"anchors": [a.to_dict() for a in self.anchors],
}
@dataclass
class AnchorSecurity:
"""One interchain anchor with the security its public network has accumulated
since the anchor was placed."""
network: str = ""
anchor_timestamp: int = 0 # unix seconds
anchor_tx_hash: str = ""
security: SecurityResult = field(default_factory=SecurityResult)
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "AnchorSecurity":
return cls(
network=d.get("network", ""),
anchor_timestamp=d.get("anchorTimestamp", 0),
anchor_tx_hash=d.get("anchorTxHash", ""),
security=SecurityResult.from_dict(d.get("security") or {}),
)
@dataclass
class HashPower:
"""Combined raw hash power across a report's PoW anchors."""
value: str = ""
units: str = ""
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "HashPower":
return cls(value=d.get("value", ""), units=d.get("units", ""))
@dataclass
class TransactionReport:
"""Per-transaction "securedBy" report: every public-chain anchor covering the
transaction's block with both raw and USD security, plus combined totals.
``hash_power`` is None when there are no PoW anchors."""
transaction_id: str = ""
prime_id: str = ""
block_id: str = ""
anchors: List[AnchorSecurity] = field(default_factory=list)
total_value_usd: str = ""
total_value_usd_formatted: str = ""
hash_power: Optional[HashPower] = None
total_normalized_score: str = ""
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "TransactionReport":
hp = d.get("hashPower")
return cls(
transaction_id=d.get("transactionId", ""),
prime_id=d.get("primeId", ""),
block_id=d.get("blockId", ""),
anchors=[AnchorSecurity.from_dict(a) for a in (d.get("anchors") or [])],
total_value_usd=d.get("totalValueUsd", ""),
total_value_usd_formatted=d.get("totalValueUsdFormatted", ""),
hash_power=HashPower.from_dict(hp) if hp else None,
total_normalized_score=d.get("totalNormalizedScore", ""),
)
@dataclass
class HealthResponse:
"""proof-measure liveness payload."""
status: str = ""
@classmethod
def from_dict(cls, d: Dict[str, Any]) -> "HealthResponse":
return cls(status=d.get("status", ""))