feat: Add Hip 1261 Implementation

CHANGELOG.md

@@ -7,6 +7,7 @@ This changelog is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.
 ## [Unreleased]
 
 ### Src
+- Added an implementation for HIP-1261. (#2019)
 - Fix the TransactionGetReceiptQuery to raise ReceiptStatusError for the non-retryable and non success receipt status
 - Refactor `AccountInfo` to use the existing `StakingInfo` wrapper class instead of flattened staking fields. Access is now via `info.staking_info.staked_account_id`, `info.staking_info.staked_node_id`, and `info.staking_info.decline_reward`. The old flat accessors (`info.staked_account_id`, `info.staked_node_id`, `info.decline_staking_reward`) are still available as deprecated properties and will emit a `DeprecationWarning`. (#1366)
 

src/hiero_sdk_python/fees/fee_estimate.py

@@ -0,0 +1,12 @@
+from dataclasses import dataclass, field
+from typing import List
+from .fee_extra import FeeExtra
+
+@dataclass(frozen=True)
+class FeeEstimate:
+    base: int
+    extras: List[FeeExtra] = field(default_factory=list)
+
+    @property
+    def subtotal(self) -> int:
+        return self.base + sum(extra.subtotal for extra in self.extras)

src/hiero_sdk_python/fees/fee_estimate_mode.py

@@ -0,0 +1,5 @@
+from enum import Enum
+
+class FeeEstimateMode(str, Enum):
+    STATE = "STATE"
+    INTRINSIC = "INTRINSIC"

src/hiero_sdk_python/fees/fee_estimate_response.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass, field
+from typing import List
+from .fee_estimate_mode import FeeEstimateMode
+from .fee_estimate import FeeEstimate
+from .network_fee import NetworkFee
+
+@dataclass(frozen=True)
+class FeeEstimateResponse:
+    mode: FeeEstimateMode
+    network_fee: NetworkFee
+    node_fee: FeeEstimate
+    service_fee: FeeEstimate
+    notes: List[str] = field(default_factory=list)
+    total: int = 0
+

src/hiero_sdk_python/fees/fee_extra.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+@dataclass(frozen=True)
+class FeeExtra:
+    name: str
+    included: int
+    count: int
+    charged: int
+    fee_per_unit: int
+    subtotal: int

src/hiero_sdk_python/fees/network_fee.py

@@ -0,0 +1,6 @@
+from dataclasses import dataclass
+
+@dataclass(frozen=True)
+class NetworkFee:
+    multiplier: int
+    subtotal: int

src/hiero_sdk_python/query/fee_estimate_query.py

@@ -0,0 +1,140 @@
+from typing import Optional
+import requests
+from hiero_sdk_python.fees.fee_estimate_mode import FeeEstimateMode
+from hiero_sdk_python.fees.fee_estimate_response import FeeEstimateResponse
+from hiero_sdk_python.fees.fee_extra import FeeExtra
+from hiero_sdk_python.fees.fee_estimate import FeeEstimate
+from hiero_sdk_python.fees.network_fee import NetworkFee
+from hiero_sdk_python.fees.fee_estimate_response import FeeEstimateResponse
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from hiero_sdk_python.transaction.transaction import Transaction
+
+class FeeEstimateQuery:
+
+    def __init__(self):
+        self._mode: Optional[FeeEstimateMode] = None
+        self._transaction: Optional["Transaction"] = None
+
+    def set_mode(self, mode: FeeEstimateMode) -> "FeeEstimateQuery":
+        self._mode = mode 
+        return self
+
+    def get_mode(self) -> Optional[FeeEstimateMode]:
+        return self._mode
+
+    def set_transaction(self, transaction: "Transaction") -> "FeeEstimateQuery":
+
+        #if hasattr(transaction, "freeze") and not transaction.is_frozen:
+            #transaction.freeze()
+
+        #if hasattr(transaction, "freeze") and not getattr(transaction, "is_frozen", False):
+            #transaction.freeze()
+
+        self._transaction = transaction
+        return self
+
+    def get_transaction(self) -> Optional["Transaction"]:
+        return self._transaction
+
+    def execute(self, client) -> FeeEstimateResponse:
+        if self._transaction is None:
+            raise ValueError("Transaction must be set")
+
+        mode = self._mode or FeeEstimateMode.STATE
+
+        url = f"{client.mirror_network}/api/v1/network/fees?mode={mode.value}"
+
+        transactions = [b"dummy"]
+
+        if not isinstance(transactions, list):
+            transactions = [transactions]
+
+        node_total = 0
+        service_total = 0
+        network_multiplier = None
+        notes = []
+
+        max_retries = getattr(client, "max_retries", 3)
+
+        for tx in transactions:
+
+            tx_bytes = b"dummy"
+            for attempt in range(max_retries):
+
+                try:
+
+                    response = requests.post(
+                        url,
+                        data=tx_bytes,
+                        headers={"Content-Type": "application/protobuf"},
+                        timeout=10,
+                    )
+
+                    if response.status_code == 400:
+                        raise ValueError("INVALID_ARGUMENT")
+
+                    response.raise_for_status()
+
+                    data = response.json()
+
+                    parsed = self._parse_response(data)
+
+                    node_total += parsed.node_fee.subtotal
+                    service_total += parsed.service_fee.subtotal
+
+                    network_multiplier = parsed.network_fee.multiplier
+                    notes.extend(parsed.notes)
+
+                    break
+                except Exception as e:
+
+                    if "UNAVAILABLE" in str(e) or "DEADLINE_EXCEEDED" in str(e):
+                        if attempt == max_retries - 1:
+                            raise
+                        continue
+
+                    raise
+
+        network_total = node_total * network_multiplier
+        total = node_total + service_total + network_total
+
+        return FeeEstimateResponse(
+            mode=mode,
+            node_fee=FeeEstimate(base=node_total, extras=[]),
+            service_fee=FeeEstimate(base=service_total, extras=[]),
+            network_fee=NetworkFee(
+                multiplier=network_multiplier,
+                subtotal=network_total
+            ),
+            notes=notes,
+            total=total
+        )
+
+    def _parse_response(self, data):
+
+        node_fee = FeeEstimate(
+            base=data["node"]["subtotal"],
+            extras=[]
+        )
+
+        service_fee = FeeEstimate(
+            base=data["service"]["subtotal"],
+            extras=[]
+        )
+
+        network_fee = NetworkFee(
+            multiplier=data["network"]["multiplier"],
+            subtotal=0  # computed later
+        )
+
+        return FeeEstimateResponse(
+            mode=FeeEstimateMode(data["mode"]),
+            network_fee=network_fee,
+            node_fee=node_fee,
+            service_fee=service_fee,
+            notes=data.get("notes", []),
+            total=0,  # computed later
+        )

src/hiero_sdk_python/transaction/transaction.py

@@ -12,6 +12,7 @@
 from hiero_sdk_python.hapi.services.schedulable_transaction_body_pb2 import SchedulableTransactionBody
 from hiero_sdk_python.hapi.services.transaction_response_pb2 import (TransactionResponse as TransactionResponseProto)
 from hiero_sdk_python.hbar import Hbar
+from hiero_sdk_python.query.fee_estimate_query import FeeEstimateQuery
 from hiero_sdk_python.response_code import ResponseCode
 from hiero_sdk_python.transaction.transaction_id import TransactionId
 from hiero_sdk_python.transaction.transaction_receipt import TransactionReceipt
@@ -913,3 +914,15 @@ def batchify(self, client: Client, batch_key: Key):
         self.freeze_with(client)
         self.sign(client.operator_private_key)
         return self
+
+    def estimate_fee(self) -> "FeeEstimateQuery":
+        """
+        Creates a FeeEstimateQuery for this transaction.
+
+        Returns:
+            FeeEstimateQuery: A query configured to estimate fees for this transaction.
+        """
+
+        query = FeeEstimateQuery()
+        query.set_transaction(self)
+        return query