hiero-sdk-python/src/hiero_sdk_python/contract/contract_info_query.py at 24472d2e8d4ac0585cf619509c6b37a5984e844d · hiero-ledger/hiero-sdk-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
"""Query to get information about a contract on the network."""

from __future__ import annotations

import logging

from hiero_sdk_python.channels import _Channel
from hiero_sdk_python.client.client import Client
from hiero_sdk_python.contract.contract_id import ContractId
from hiero_sdk_python.contract.contract_info import ContractInfo
from hiero_sdk_python.executable import _Method
from hiero_sdk_python.hapi.services import (
    contract_get_info_pb2,
    query_pb2,
    response_pb2,
)
from hiero_sdk_python.hapi.services.contract_get_info_pb2 import ContractGetInfoResponse
from hiero_sdk_python.query.query import Query


logger = logging.getLogger(__name__)


class ContractInfoQuery(Query):
    """
    A query to retrieve information about a specific Contract.

    This class constructs and executes a query to retrieve information
    about a contract on the network, including the contract's properties and settings.

    """

    def __init__(self, contract_id: ContractId | None = None) -> None:
        """
        Initializes a new ContractInfoQuery instance with an optional contract_id.

        Args:
            contract_id (ContractId, optional): The ID of the contract to query.
        """
        super().__init__()
        self.contract_id: ContractId | None = contract_id

    def set_contract_id(self, contract_id: ContractId | None) -> ContractInfoQuery:
        """
        Sets the ID of the contract to query.

        Args:
            contract_id (ContractId | None): The ID of the contract.
        """
        self.contract_id = contract_id
        return self

    def _make_request(self) -> query_pb2.Query:
        """
        Constructs the protobuf request for the query.

        Builds a ContractGetInfoQuery protobuf message with the
        appropriate header and contract ID.

        Returns:
            Query: The protobuf query message.

        Raises:
            ValueError: If the contract ID is not set.
            Exception: If any other error occurs during request construction.
        """
        try:
            if not self.contract_id:
                raise ValueError("Contract ID must be set before making the request.")

            query_header = self._make_request_header()

            contract_info_query = contract_get_info_pb2.ContractGetInfoQuery()
            contract_info_query.header.CopyFrom(query_header)
            contract_info_query.contractID.CopyFrom(self.contract_id._to_proto())

            query = query_pb2.Query()
            query.contractGetInfo.CopyFrom(contract_info_query)

            return query
        except Exception as e:
            logger.error("Exception in _make_request: %s", e, exc_info=True)
            raise

    def _get_method(self, channel: _Channel) -> _Method:
        """
        Returns the appropriate gRPC method for the contract info query.

        Implements the abstract method from Query to provide the specific
        gRPC method for getting contract information.

        Args:
            channel (_Channel): The channel containing service stubs

        Returns:
            _Method: The method wrapper containing the query function
        """
        return _Method(transaction_func=None, query_func=channel.smart_contract.getContractInfo)

    def execute(self, client: Client, timeout: int | float | None = None) -> ContractInfo:
        """
        Executes the contract info query.

        Sends the query to the network and processes the response
        to return a ContractInfo object.

        This function delegates the core logic to `_execute()`, and may propagate
        exceptions raised by it.

        Args:
            client (Client): The client instance to use for execution
            timeout (int | float, optional): The total execution timeout (in seconds) for this execution.

        Returns:
            ContractInfo: The contract info from the network

        Raises:
            PrecheckError: If the query fails with a non-retryable error
            MaxAttemptsError: If the query fails after the maximum number of attempts
            ReceiptStatusError: If the query fails with a receipt status error
        """
        self._before_execute(client)
        response = self._execute(client, timeout)

        return ContractInfo._from_proto(response.contractGetInfo.contractInfo)

    def _get_query_response(self, response: response_pb2.Response) -> ContractGetInfoResponse.ContractInfo:
        """
        Extracts the contract info response from the full response.

        Implements the abstract method from Query to extract the
        specific contract info response object.

        Args:
            response: The full response from the network

        Returns:
            The contract get info response object
        """
        return response.contractGetInfo