Bases: Base
Query Catalog
The query catalog contains one CatalogEntry object for each valid query in the Tellor network.
It is stored as a mapping of query names (i.e. tags) to CatalogEntry objects.
Source code in telliot_feeds/queries/catalog.py
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 | @dataclass
class Catalog(Base):
"""Query Catalog
The query catalog contains one `CatalogEntry` object for each valid query in the Tellor network.
It is stored as a mapping of query names (i.e. tags) to `CatalogEntry` objects.
"""
_entries: Dict[str, CatalogEntry] = field(default_factory=dict)
def add_entry(self, tag: str, title: str, q: OracleQuery, active: bool = True) -> None:
"""Add a new entry to the catalog."""
if tag in self._entries:
raise Exception(f"Error adding query entry: {tag} already exists")
if isinstance(q, AbiQuery):
abi = json.dumps(q.abi)
else:
abi = ""
entry = CatalogEntry(
tag=tag,
title=title,
query_type=q.__class__.__name__,
descriptor=q.descriptor,
query_id=f"0x{q.query_id.hex()}",
active=active,
abi=abi,
)
self._entries[tag] = entry
def find(
self,
*,
tag: Optional[str] = None,
query_id: Optional[str] = None,
query_type: Optional[str] = None,
active: Optional[bool] = None,
) -> List[OracleQuery]:
"""Search the query catalog for matching entries."""
entries = []
for entry in self._entries.values():
if tag is not None:
if tag not in entry.tag: # includes search for substring
continue
if query_id is not None:
# Add 0x if necessary for match
if query_id[:2] not in ["0x", "0X"]:
query_id = "0x" + query_id
if query_id.lower() != entry.query_id.lower():
continue
if query_type is not None:
if query_type.lower() != entry.query_type.lower():
continue
if active is not None:
if active != entry.active:
continue
entries.append(entry)
return entries
def to_yaml(self) -> str:
all_entries = self.find()
return yaml.dump(clamfig.serialize(all_entries), sort_keys=False) # type: ignore
def to_markdown(self) -> str:
lines = ["# TellorX Query Catalog", ""]
for entry in self.find():
lines.append(f"## {entry.title}")
lines.append("")
lines.append("| Parameter | Value |")
lines.append("| --- | --- |")
lines.append(f"| Tag | `{entry.tag}` |")
lines.append(f"| Active | `{entry.active}` |")
lines.append(f"| Type | `{entry.query_type}` |")
lines.append(f"| Descriptor | `{entry.descriptor}` |")
lines.append(f"| Encoding ABI | `{entry.abi}` |")
lines.append(f"| Query ID | `{entry.query_id}` |") # type: ignore
lines.append(f"| Query data | `0x{entry.query.query_data.hex()}` |")
lines.append("")
return "\n".join(lines)
|
add_entry(tag, title, q, active=True)
Add a new entry to the catalog.
Source code in telliot_feeds/queries/catalog.py
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70 | def add_entry(self, tag: str, title: str, q: OracleQuery, active: bool = True) -> None:
"""Add a new entry to the catalog."""
if tag in self._entries:
raise Exception(f"Error adding query entry: {tag} already exists")
if isinstance(q, AbiQuery):
abi = json.dumps(q.abi)
else:
abi = ""
entry = CatalogEntry(
tag=tag,
title=title,
query_type=q.__class__.__name__,
descriptor=q.descriptor,
query_id=f"0x{q.query_id.hex()}",
active=active,
abi=abi,
)
self._entries[tag] = entry
|
find(*, tag=None, query_id=None, query_type=None, active=None)
Search the query catalog for matching entries.
Source code in telliot_feeds/queries/catalog.py
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 | def find(
self,
*,
tag: Optional[str] = None,
query_id: Optional[str] = None,
query_type: Optional[str] = None,
active: Optional[bool] = None,
) -> List[OracleQuery]:
"""Search the query catalog for matching entries."""
entries = []
for entry in self._entries.values():
if tag is not None:
if tag not in entry.tag: # includes search for substring
continue
if query_id is not None:
# Add 0x if necessary for match
if query_id[:2] not in ["0x", "0X"]:
query_id = "0x" + query_id
if query_id.lower() != entry.query_id.lower():
continue
if query_type is not None:
if query_type.lower() != entry.query_type.lower():
continue
if active is not None:
if active != entry.active:
continue
entries.append(entry)
return entries
|