Skip to content
Open
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,7 @@ This changelog is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.
- Added first-class support for EVM address aliases in `AccountId`, including parsing, serialization, Mirror Node population helpers.
- Add automated bot to recommend next issues to contributors after their first PR merge (#1380)
- Added dry-run support and refactored `.github/workflows/bot-workflows.yml` to use dedicated script `.github/scripts/bot-workflows.js` for improved maintainability and testability. (`#1288`)
- Type hints to exception classes (`PrecheckError`, `MaxAttemptsError`, `ReceiptStatusError`) constructors and string methods.

### Changed
- Updated GitHub Actions setup-node action to v6.2.0.
Expand Down
58 changes: 58 additions & 0 deletions examples/errors/max_attempts_error.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
#!/usr/bin/env python3
"""
Example demonstrating how to handle MaxAttemptsError in the Hiero SDK.

run:
uv run examples/errors/max_attempts_error.py
python examples/errors/max_attempts_error.py
"""

from hiero_sdk_python import (
Client,
TransactionGetReceiptQuery,
TransactionId,
)
from hiero_sdk_python.exceptions import MaxAttemptsError

def main() -> None:
# Initialize the client
client = Client.from_env()
operator_id = client.operator_account_id

# Configure client to fail quickly
# This sets the maximum number of attempts for any request to 1
client.max_attempts = 1

print("Attempting to fetch receipt with restricted max attempts...")

# We generate a random transaction ID that definitely doesn't exist.
# The network would normally return RECEIPT_NOT_FOUND, but depending on the
# node's state or if we simulate a network blip, the SDK's retry mechanism kicks in.
# By forcing max_attempts=1, we prevent retries.
# Note: Triggering a pure MaxAttemptsError usually requires a timeout or busy node.
# This example demonstrates the structure of handling the error.

# Using a generated TransactionId
tx_id = TransactionId.generate(operator_id)

try:
TransactionGetReceiptQuery().set_transaction_id(tx_id).execute(client)
print("Query finished (unexpected for this example test).")

except MaxAttemptsError as e:
print("\nCaught MaxAttemptsError!")
print(f"Node ID: {e.node_id}")
print(f"Message: {e.message}")
print("This error means the SDK gave up after reaching the maximum number of retry attempts.")

except Exception as e:
# Note: In a real network test with a made-up ID, we might get ReceiptStatusError
# or PrecheckError (RECEIPT_NOT_FOUND). MaxAttemptsError typically happens
# on network timeouts or BUSY responses.
print(f"\nCaught unexpected error (expected for this specific simulation): {type(e).__name__}")
print(f"Details: {e}")
print("\n(To verify MaxAttemptsError logic, this example relies on the client's retry configuration)")


if __name__ == "__main__":
main()
61 changes: 61 additions & 0 deletions examples/errors/precheck_error.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
#!/usr/bin/env python3
"""
Example demonstrating how to handle PrecheckError in the Hiero SDK.

run:
uv run examples/errors/precheck_error.py
python examples/errors/precheck_error.py
"""

from hiero_sdk_python.exceptions import PrecheckError
from hiero_sdk_python import (
Client,
TransferTransaction,
ResponseCode,
AccountId
)

def main() -> None:
# Initialize the client
client = Client.from_env()
operator_id = client.operator_account_id
operator_key = client.operator_private_key

print("Creating transaction with invalid parameters to force PrecheckError...")

# Create a simple transfer transaction
# To trigger a PrecheckError, we set the transaction valid duration to 0.
# The node's precheck validation requires a valid duration, so this will fail immediately.
transaction = (
TransferTransaction()
.add_hbar_transfer(operator_id, -1)
.add_hbar_transfer(AccountId(0, 0, 3), 1)
)

# Set the invalid duration directly on the attribute
transaction.transaction_valid_duration = 0

transaction = (
transaction
.freeze_with(client)
.sign(operator_key)
)


try:
print("Executing transaction...")
transaction.execute(client)
print("Transaction unexpectedly succeeded (this should not happen).")

except PrecheckError as e:
print("\nCaught PrecheckError!")
# This should print: Status: INVALID_TRANSACTION_DURATION (ResponseCode)
print(f"Status: {e.status} ({ResponseCode(e.status).name})")
print(f"Transaction ID: {e.transaction_id}")
print("This error means the transaction failed validation at the node *before* reaching consensus.")

except Exception as e:
print(f"\nAn unexpected error occurred: {type(e).__name__}: {e}")

if __name__ == "__main__":
main()
49 changes: 16 additions & 33 deletions examples/errors/receipt_status_error.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,41 +6,20 @@
uv run examples/errors/receipt_status_error.py
python examples/errors/receipt_status_error.py
"""
import os
import dotenv

from hiero_sdk_python.client.client import Client
from hiero_sdk_python.account.account_id import AccountId
from hiero_sdk_python.crypto.private_key import PrivateKey
from hiero_sdk_python.response_code import ResponseCode
from hiero_sdk_python.tokens.token_associate_transaction import (
from hiero_sdk_python import (
Client,
ResponseCode,
TokenAssociateTransaction,
TokenId
)
from hiero_sdk_python.tokens.token_id import TokenId
from hiero_sdk_python.exceptions import ReceiptStatusError

dotenv.load_dotenv()


def main():
# Initialize the client
# For this example, we assume we are running against a local node or testnet
# You would typically load these from environment variables
operator_id_str = os.environ.get("OPERATOR_ID", "")
operator_key_str = os.environ.get("OPERATOR_KEY", "")

try:
operator_id = AccountId.from_string(operator_id_str)
operator_key = PrivateKey.from_string(operator_key_str)
except Exception as e:
print(f"Error parsing operator credentials: {e}")
return

client = Client()
client.set_operator(operator_id, operator_key)

# Create a transaction that is likely to fail post-consensus
# Here we try to associate a non-existent token
def main() -> None:
# Client.from_env() automatically loads .env and sets up the operator
client = Client.from_env()

operator_id = client.operator_account_id
operator_key = client.operator_private_key

print("Creating transaction...")
transaction = (
Expand All @@ -60,9 +39,13 @@ def main():

# Check if the execution raised something other than SUCCESS
# If not, we raise our custom ReceiptStatusError for handling.
if receipt.status is None:
raise ValueError("Receipt missing status")
if receipt.status != ResponseCode.SUCCESS:
if not receipt.transaction_id:
raise ValueError("Receipt missing transaction_id; cannot raise ReceiptStatusError")
raise ReceiptStatusError(receipt.status, receipt.transaction_id, receipt)
# If we reach here, the transaction succeeded

print("Transaction successful!")

# This exception is raised when the transaction raised something other than SUCCESS
Expand All @@ -80,4 +63,4 @@ def main():


if __name__ == "__main__":
main()
main()
80 changes: 51 additions & 29 deletions src/hiero_sdk_python/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,14 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Optional

from hiero_sdk_python.response_code import ResponseCode

if TYPE_CHECKING:
from hiero_sdk_python import (
TransactionId,
TransactionReceipt
)

class PrecheckError(Exception):
"""
Exception thrown when a transaction fails its precheck validation.
Expand All @@ -11,81 +20,94 @@ class PrecheckError(Exception):
transaction_id (TransactionId): The ID of the transaction that failed.
message (str): The message of the error. If not provided, a default message is generated.
"""
def __init__(self, status, transaction_id=None, message=None):

def __init__(
self,
status: ResponseCode,
transaction_id: Optional[TransactionId] = None,
message: Optional[str] = None,
) -> None:
self.status = status
self.transaction_id = transaction_id

# Build a default message if none provided
if message is None:
status_name = ResponseCode(status).name
message = f"Transaction failed precheck with status: {status_name} ({status})"
if transaction_id:
message += f", transaction ID: {transaction_id}"

self.message = message
super().__init__(self.message)
def __str__(self):

def __str__(self) -> str:
return self.message
def __repr__(self):

def __repr__(self) -> str:
return f"PrecheckError(status={self.status}, transaction_id={self.transaction_id})"


class MaxAttemptsError(Exception):
"""
Exception raised when the maximum number of attempts for a request has been reached.

Attributes:
message (str): The error message explaining why the maximum attempts were reached
node_id (str): The ID of the node that was being contacted when the max attempts were reached
last_error (Exception): The last error that occurred during the final attempt
"""
def __init__(self, message, node_id, last_error=None):

def __init__(self, message: str, node_id: str, last_error: Optional[Exception] = None) -> None:
self.node_id = node_id
self.last_error = last_error

# Build a comprehensive error message
error_message = message
if last_error is not None:
error_message += f"; last error: {str(last_error)}"

self.message = error_message
super().__init__(self.message)
def __str__(self):

def __str__(self) -> str:
return self.message
def __repr__(self):

def __repr__(self) -> str:
return f"MaxAttemptsError(message='{self.message}', node_id='{self.node_id}')"



class ReceiptStatusError(Exception):
"""
Exception raised when a transaction receipt contains an error status.

Attributes:
status (ResponseCode): The error status code from the receipt
transaction_id (TransactionId): The ID of the transaction that failed
transaction_receipt (TransactionReceipt): The receipt containing the error status
message (str): The error message describing the failure
"""

def __init__(self, status, transaction_id, transaction_receipt, message=None):

def __init__(
self,
status: ResponseCode,
transaction_id: TransactionId,
transaction_receipt: TransactionReceipt,
message: Optional[str] = None,
) -> None:
self.status = status
self.transaction_id = transaction_id
self.transaction_receipt = transaction_receipt

# Build a default message if none provided
if message is None:
status_name = ResponseCode(status).name
message = f"Receipt for transaction {transaction_id} contained error status: {status_name} ({status})"

self.message = message
super().__init__(self.message)
def __str__(self):

def __str__(self) -> str:
return self.message
def __repr__(self):
return f"ReceiptStatusError(status={self.status}, transaction_id={self.transaction_id})"

def __repr__(self) -> str:
return f"ReceiptStatusError(status={self.status}, transaction_id={self.transaction_id})"
53 changes: 53 additions & 0 deletions tests/unit/exceptions_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
import pytest
from unittest.mock import Mock
from hiero_sdk_python.exceptions import PrecheckError, MaxAttemptsError, ReceiptStatusError
from hiero_sdk_python.response_code import ResponseCode

pytestmark = pytest.mark.unit

def test_precheck_error_typing_and_defaults():
"""Test PrecheckError with and without optional arguments."""
# Mock TransactionId
tx_id_mock = Mock()
tx_id_mock.__str__ = Mock(return_value="[email protected]")

# Case 1: All arguments provided
err = PrecheckError(ResponseCode.INVALID_TRANSACTION, tx_id_mock, "Custom error")
assert err.status == ResponseCode.INVALID_TRANSACTION
assert err.transaction_id is tx_id_mock
assert str(err) == "Custom error"
assert repr(err) == f"PrecheckError(status={ResponseCode.INVALID_TRANSACTION}, transaction_id={tx_id_mock})"

# Case 2: Default message generation
err_default = PrecheckError(ResponseCode.INVALID_TRANSACTION, tx_id_mock)
expected_msg = "Transaction failed precheck with status: INVALID_TRANSACTION (1), transaction ID: [email protected]"
assert str(err_default) == expected_msg

def test_max_attempts_error_typing():
"""Test MaxAttemptsError with required and optional arguments."""
# Case 1: With last_error
inner_error = ValueError("Connection failed")
err = MaxAttemptsError("Max attempts reached", "0.0.3", inner_error)
assert err.node_id == "0.0.3"
assert err.last_error is inner_error
assert "Max attempts reached" in str(err)
assert "Connection failed" in str(err)

# Case 2: Without last_error
err_simple = MaxAttemptsError("Just failed", "0.0.4")
assert str(err_simple) == "Just failed"

def test_receipt_status_error_typing():
"""Test ReceiptStatusError initialization."""
tx_id_mock = Mock()
receipt_mock = Mock()

# Case 1: Default message
err = ReceiptStatusError(ResponseCode.RECEIPT_NOT_FOUND, tx_id_mock, receipt_mock)
assert err.status == ResponseCode.RECEIPT_NOT_FOUND
assert err.transaction_receipt is receipt_mock
assert "RECEIPT_NOT_FOUND" in str(err)

# Case 2: Custom message
err_custom = ReceiptStatusError(ResponseCode.FAIL_INVALID, tx_id_mock, receipt_mock, "Fatal receipt error")
assert str(err_custom) == "Fatal receipt error"