Skip to content
Open
Show file tree
Hide file tree
Changes from all 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 @@ -134,6 +134,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.
- Added `/working` command to reset the inactivity timer on issues and PRs. ([#1552](https://github.com/hiero-ledger/hiero-sdk-python/issues/1552))

### Changed
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"