Skip to content

Query Base Classes

OracleQuery

Bases: Serializable

Oracle Query

An OracleQuery specifies how to pose a question to the Tellor Oracle and how to format/interpret the response.

The OracleQuery class serves as the base class for all Queries, and implements default behaviors. Each subclass corresponds to a unique Query Type supported by the TellorX network.

All public attributes of an OracleQuery represent a parameter that can be used to customize the query.

The base class provides:

  • Generation of the query descriptor JSON string. This string provides a simple platform and language independent way to identify a query.

  • Calculation of the id field from query_data. This value is used for the TellorX.Oracle.tipQuery() and TellorX.Oracle.submitValue() contract calls.

Subclasses must provide:

  • Encoding of the descriptor string to compute the query_data attribute, which is used for the data field of a TellorX.Oracle.tipQuery() contract call.
Source code in telliot_feeds/queries/query.py 
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
class OracleQuery(Serializable):
    """Oracle Query

    An OracleQuery specifies how to pose a question to the
    Tellor Oracle and how to format/interpret the response.

    The OracleQuery class serves
    as the base class for all Queries, and implements default behaviors.
    Each subclass corresponds to a unique Query Type supported
    by the TellorX network.

    All public attributes of an OracleQuery represent a parameter that can
    be used to customize the query.

    The base class provides:

    - Generation of the query `descriptor` JSON string.
    This string provides a simple platform and language independent
    way to identify a query.

    - Calculation of the `id` field from `query_data`.  This value is used for the
      `TellorX.Oracle.tipQuery()` and `TellorX.Oracle.submitValue()`
      contract calls.

    Subclasses must provide:

    - Encoding of the `descriptor` string to compute the `query_data` attribute,
    which is used for the `data` field of a `TellorX.Oracle.tipQuery()` contract call.

    """

    @property
    def value_type(self) -> ValueType:
        """Returns the ValueType expected by the current Query configuration

        The value type defines required data type/structure of the
        ``value`` submitted to the contract through
        ``TellorX.Oracle.submitValue()``

        This method *must* be implemented by subclasses
        """
        raise NotImplementedError

    @property
    def descriptor(self) -> str:
        """Get the query descriptor string.

        The Query descriptor is a unique string representation of the query, including
        all parameter values.  The string must be in valid JSON format (http://www.json.org).

        """
        state = self.get_state()
        json_str = json.dumps(state, separators=(",", ":"))
        return json_str

    @property
    def query_id(self) -> bytes:
        """Returns the query ``id`` for use with the
        ``TellorX.Oracle.tipQuery()`` and ``TellorX.Oracle.submitValue()``
        contract calls.
        """
        return bytes(Web3.keccak(self.query_data))

    @property
    def query_data(self) -> bytes:
        """Encode the query `descriptor` to create the query `data` field for
        use in the ``TellorX.Oracle.tipQuery()`` contract call.

        This method *must* be implemented by subclasses
        """
        raise NotImplementedError

    @staticmethod
    def get_query_from_data(query_data: bytes) -> Optional[OracleQuery]:
        """Recreate an oracle query from `query_data`"""
        raise NotImplementedError

value_type: ValueType property

Returns the ValueType expected by the current Query configuration

The value type defines required data type/structure of the value submitted to the contract through TellorX.Oracle.submitValue()

This method must be implemented by subclasses

descriptor: str property

Get the query descriptor string.

The Query descriptor is a unique string representation of the query, including all parameter values. The string must be in valid JSON format (http://www.json.org).

query_id: bytes property

Returns the query id for use with the TellorX.Oracle.tipQuery() and TellorX.Oracle.submitValue() contract calls.

query_data: bytes property

Encode the query descriptor to create the query data field for use in the TellorX.Oracle.tipQuery() contract call.

This method must be implemented by subclasses

get_query_from_data(query_data) staticmethod

Recreate an oracle query from query_data

Source code in telliot_feeds/queries/query.py 
90
91
92
93
@staticmethod
def get_query_from_data(query_data: bytes) -> Optional[OracleQuery]:
    """Recreate an oracle query from `query_data`"""
    raise NotImplementedError

JsonQuery

Bases: OracleQuery

An Oracle Query that uses JSON-encoding to compute the query_data.

Source code in telliot_feeds/queries/json_query.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
class JsonQuery(OracleQuery):
    """An Oracle Query that uses JSON-encoding to compute the query_data."""

    @property
    def query_data(self) -> bytes:
        """Encode the query `descriptor` to create the query `data` field for
        use in the ``TellorX.Oracle.tipQuery()`` contract call.

        """
        return self.descriptor.encode("utf-8")

    @staticmethod
    def get_query_from_data(query_data: bytes) -> OracleQuery:
        """Recreate an oracle query from `query_data`"""
        descriptor = query_data.decode("utf-8")

        return query_from_descriptor(descriptor)

query_data: bytes property

Encode the query descriptor to create the query data field for use in the TellorX.Oracle.tipQuery() contract call.

get_query_from_data(query_data) staticmethod

Recreate an oracle query from query_data

Source code in telliot_feeds/queries/json_query.py 
16
17
18
19
20
21
@staticmethod
def get_query_from_data(query_data: bytes) -> OracleQuery:
    """Recreate an oracle query from `query_data`"""
    descriptor = query_data.decode("utf-8")

    return query_from_descriptor(descriptor)

AbiQuery

Bases: OracleQuery

An Oracle Query that uses ABI-encoding to compute the query_data.

Attributes:

Name Type Description
abi ClassVar[list[dict[str, str]]]

The ABI used for encoding/decoding parameters. Each subclass must defind the ABI. The ABI is an ordered list, with one entry for each query parameter. Each parameter should include a dict with two entries: {"name": , "type": } Parameter types must be valid solidity ABI type string. See https://docs.soliditylang.org/en/develop/types.html for reference.
Source code in telliot_feeds/queries/abi_query.py 
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
class AbiQuery(OracleQuery):
    """An Oracle Query that uses ABI-encoding to compute the query_data.

    Attributes:
        abi:
            The ABI used for encoding/decoding parameters.
            Each subclass must defind the ABI.
            The ABI is an ordered list, with one entry for each query parameter.
            Each parameter should include a dict with two entries:
                {"name": <parameter name>, "type": <parameter type>}
            Parameter types must be valid solidity ABI type string.
                See https://docs.soliditylang.org/en/develop/types.html for reference.

    """

    abi: ClassVar[list[dict[str, str]]] = []

    @property
    def query_data(self) -> bytes:
        """Encode the query type and parameters to create the query data.

        This method uses ABI encoding to encode the query's parameter values.
        """
        # If the query has parameters
        if self.abi:
            param_values = [getattr(self, p["name"]) for p in self.abi]
            param_types = [p["type"] for p in self.abi]
            encoded_params = encode_abi(param_types, param_values)

        # If the query has no real parameters, and only the default "phantom" parameter
        else:
            # By default, the queries with no real parameters have a phantom parameter with
            # a consistent value of empty bytes. The encoding of these empty bytese in
            # Python does not match the encoding in Solidity, so the bytes are generated
            # manually like so:
            left_side = b"\0 ".rjust(32, b"\0")
            right_side = b"\0".rjust(32, b"\0")
            encoded_params = left_side + right_side

        return encode_abi(["string", "bytes"], [type(self).__name__, encoded_params])

    @staticmethod
    def get_query_from_data(query_data: bytes) -> Optional[OracleQuery]:
        """Recreate an oracle query from the `query_data` field"""
        try:
            query_type, encoded_param_values = decode_abi(["string", "bytes"], query_data)
        except OverflowError:
            logger.error("OverflowError while decoding query data.")
            return None
        try:
            cls = Registry.registry[query_type]
        except KeyError:
            logger.error(f"Unsupported query type: {query_type}")
            return None
        params_abi = cls.abi
        param_names = [p["name"] for p in params_abi]
        param_types = [p["type"] for p in params_abi]
        param_values = decode_abi(param_types, encoded_param_values)

        params = dict(zip(param_names, param_values))

        return deserialize({"type": query_type, **params})  # type: ignore

query_data: bytes property

Encode the query type and parameters to create the query data.

This method uses ABI encoding to encode the query's parameter values.

get_query_from_data(query_data) staticmethod

Recreate an oracle query from the query_data field

Source code in telliot_feeds/queries/abi_query.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@staticmethod
def get_query_from_data(query_data: bytes) -> Optional[OracleQuery]:
    """Recreate an oracle query from the `query_data` field"""
    try:
        query_type, encoded_param_values = decode_abi(["string", "bytes"], query_data)
    except OverflowError:
        logger.error("OverflowError while decoding query data.")
        return None
    try:
        cls = Registry.registry[query_type]
    except KeyError:
        logger.error(f"Unsupported query type: {query_type}")
        return None
    params_abi = cls.abi
    param_names = [p["name"] for p in params_abi]
    param_types = [p["type"] for p in params_abi]
    param_values = decode_abi(param_types, encoded_param_values)

    params = dict(zip(param_names, param_values))

    return deserialize({"type": query_type, **params})  # type: ignore