Changeset View
Changeset View
Standalone View
Standalone View
test/functional/test_framework/chronik/client.py
| Show First 20 Lines • Show All 42 Lines • ▼ Show 20 Lines | |||||
| class ChronikScriptClient: | class ChronikScriptClient: | ||||
| def __init__(self, client: 'ChronikClient', script_type: str, | def __init__(self, client: 'ChronikClient', script_type: str, | ||||
| script_payload: str) -> None: | script_payload: str) -> None: | ||||
| self.client = client | self.client = client | ||||
| self.script_type = script_type | self.script_type = script_type | ||||
| self.script_payload = script_payload | self.script_payload = script_payload | ||||
| def _query_params(self, page=None, page_size=None) -> str: | |||||
| if page is not None and page_size is not None: | |||||
| return f'?page={page}&page_size={page_size}' | |||||
| elif page is not None: | |||||
| return f'?page={page}' | |||||
| elif page_size is not None: | |||||
| return f'?page_size={page_size}' | |||||
| else: | |||||
| return '' | |||||
| def confirmed_txs(self, page=None, page_size=None): | def confirmed_txs(self, page=None, page_size=None): | ||||
| query = self._query_params(page, page_size) | query = _page_query_params(page, page_size) | ||||
| return self.client._request_get( | return self.client._request_get( | ||||
| f'/script/{self.script_type}/{self.script_payload}/confirmed-txs{query}', | f'/script/{self.script_type}/{self.script_payload}/confirmed-txs{query}', | ||||
| pb.TxHistoryPage) | pb.TxHistoryPage) | ||||
| def history(self, page=None, page_size=None): | def history(self, page=None, page_size=None): | ||||
| query = self._query_params(page, page_size) | query = _page_query_params(page, page_size) | ||||
| return self.client._request_get( | return self.client._request_get( | ||||
| f'/script/{self.script_type}/{self.script_payload}/history{query}', | f'/script/{self.script_type}/{self.script_payload}/history{query}', | ||||
| pb.TxHistoryPage) | pb.TxHistoryPage) | ||||
| def unconfirmed_txs(self): | def unconfirmed_txs(self): | ||||
| return self.client._request_get( | return self.client._request_get( | ||||
| f'/script/{self.script_type}/{self.script_payload}/unconfirmed-txs', | f'/script/{self.script_type}/{self.script_payload}/unconfirmed-txs', | ||||
| pb.TxHistoryPage) | pb.TxHistoryPage) | ||||
| ▲ Show 20 Lines • Show All 63 Lines • ▼ Show 20 Lines | def _request_get(self, path: str, pb_type): | ||||
| return ChronikResponse(response.status, ok_proto=ok_proto) | return ChronikResponse(response.status, ok_proto=ok_proto) | ||||
| def blockchain_info(self) -> ChronikResponse: | def blockchain_info(self) -> ChronikResponse: | ||||
| return self._request_get('/blockchain-info', pb.BlockchainInfo) | return self._request_get('/blockchain-info', pb.BlockchainInfo) | ||||
| def block(self, hash_or_height: Union[str, int]) -> ChronikResponse: | def block(self, hash_or_height: Union[str, int]) -> ChronikResponse: | ||||
| return self._request_get(f'/block/{hash_or_height}', pb.Block) | return self._request_get(f'/block/{hash_or_height}', pb.Block) | ||||
| def block_txs(self, hash_or_height: Union[str, int], | |||||
| page=None, page_size=None) -> ChronikResponse: | |||||
| query = _page_query_params(page, page_size) | |||||
| return self._request_get( | |||||
| f'/block-txs/{hash_or_height}{query}', pb.TxHistoryPage) | |||||
| def blocks(self, start_height: int, end_height: int) -> ChronikResponse: | def blocks(self, start_height: int, end_height: int) -> ChronikResponse: | ||||
| return self._request_get(f'/blocks/{start_height}/{end_height}', pb.Blocks) | return self._request_get(f'/blocks/{start_height}/{end_height}', pb.Blocks) | ||||
| def tx(self, txid: str) -> ChronikResponse: | def tx(self, txid: str) -> ChronikResponse: | ||||
| return self._request_get(f'/tx/{txid}', pb.Tx) | return self._request_get(f'/tx/{txid}', pb.Tx) | ||||
| def raw_tx(self, txid: str) -> bytes: | def raw_tx(self, txid: str) -> bytes: | ||||
| return self._request_get(f'/raw-tx/{txid}', pb.RawTx) | return self._request_get(f'/raw-tx/{txid}', pb.RawTx) | ||||
| def script(self, script_type: str, script_payload: str) -> ChronikScriptClient: | def script(self, script_type: str, script_payload: str) -> ChronikScriptClient: | ||||
| return ChronikScriptClient(self, script_type, script_payload) | return ChronikScriptClient(self, script_type, script_payload) | ||||
| def ws(self, *, timeout=None) -> ChronikWs: | def ws(self, *, timeout=None) -> ChronikWs: | ||||
| ws = websocket.WebSocket() | ws = websocket.WebSocket() | ||||
| ws.connect(f'ws://{self.host}:{self.port}/ws', timeout=timeout) | ws.connect(f'ws://{self.host}:{self.port}/ws', timeout=timeout) | ||||
| return ChronikWs(ws) | return ChronikWs(ws) | ||||
| def _page_query_params(page=None, page_size=None) -> str: | |||||
| if page is not None and page_size is not None: | |||||
| return f'?page={page}&page_size={page_size}' | |||||
| elif page is not None: | |||||
| return f'?page={page}' | |||||
| elif page_size is not None: | |||||
| return f'?page_size={page_size}' | |||||
| else: | |||||
| return '' | |||||