1263745490

Build Type

push

github

Committed by Lex Berezhny

Commit Message

v0.104.0

Run Details

2863 of 8167 branches covered (35.06%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

3 existing lines in 2 files now uncovered.

19484 of 24672 relevant lines covered (78.97%)

3.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

58.97

/lbry/wallet/server/daemon.py

import asyncio
import itertools
import json
import time
from functools import wraps

import aiohttp
from prometheus_client import Gauge, Histogram
from lbry.utils import LRUCacheWithMetrics
from lbry.wallet.rpc.jsonrpc import RPCError
from lbry.wallet.server.util import hex_to_bytes, class_logger
from lbry.wallet.rpc import JSONRPC


class DaemonError(Exception):
    """Raised when the daemon returns an error in its results."""


class WarmingUpError(Exception):
    """Internal - when the daemon is warming up."""


class WorkQueueFullError(Exception):
    """Internal - when the daemon's work queue is full."""


NAMESPACE = "wallet_server"


class Daemon:
    """Handles connections to a daemon at the given URL."""

    WARMING_UP = -28
    id_counter = itertools.count()

    lbrycrd_request_time_metric = Histogram(
        "lbrycrd_request", "lbrycrd requests count", namespace=NAMESPACE, labelnames=("method",)
    )
    lbrycrd_pending_count_metric = Gauge(
        "lbrycrd_pending_count", "Number of lbrycrd rpcs that are in flight", namespace=NAMESPACE,
        labelnames=("method",)
    )

    def __init__(self, coin, url, max_workqueue=10, init_retry=0.25,
                 max_retry=4.0):
        self.coin = coin
        self.logger = class_logger(__name__, self.__class__.__name__)
        self.set_url(url)
        # Limit concurrent RPC calls to this number.
        # See DEFAULT_HTTP_WORKQUEUE in bitcoind, which is typically 16
        self.workqueue_semaphore = asyncio.Semaphore(value=max_workqueue)
        self.init_retry = init_retry
        self.max_retry = max_retry
        self._height = None
        self.available_rpcs = {}
        self.connector = aiohttp.TCPConnector()
        self._block_hash_cache = LRUCacheWithMetrics(100000)
        self._block_cache = LRUCacheWithMetrics(2 ** 16, metric_name='block', namespace=NAMESPACE)

    async def close(self):
        if self.connector:
            await self.connector.close()
            self.connector = None

    def set_url(self, url):
        """Set the URLS to the given list, and switch to the first one."""
        urls = url.split(',')
        urls = [self.coin.sanitize_url(url) for url in urls]
        for n, url in enumerate(urls):
            status = '' if n else ' (current)'
            logged_url = self.logged_url(url)
            self.logger.info(f'daemon #{n + 1} at {logged_url}{status}')
        self.url_index = 0
        self.urls = urls

    def current_url(self):
        """Returns the current daemon URL."""
        return self.urls[self.url_index]

    def logged_url(self, url=None):
        """The host and port part, for logging."""
        url = url or self.current_url()
        return url[url.rindex('@') + 1:]

    def failover(self):
        """Call to fail-over to the next daemon URL.

        Returns False if there is only one, otherwise True.
        """
        if len(self.urls) > 1:
            self.url_index = (self.url_index + 1) % len(self.urls)
            self.logger.info(f'failing over to {self.logged_url()}')
            return True
        return False

    def client_session(self):
        """An aiohttp client session."""
        return aiohttp.ClientSession(connector=self.connector, connector_owner=False)

    async def _send_data(self, data):
        if not self.connector:
            raise asyncio.CancelledError('Tried to send request during shutdown.')
        async with self.workqueue_semaphore:
            async with self.client_session() as session:
                async with session.post(self.current_url(), data=data) as resp:
                    kind = resp.headers.get('Content-Type', None)
                    if kind == 'application/json':
                        return await resp.json()
                    # bitcoind's HTTP protocol "handling" is a bad joke
                    text = await resp.text()
                    if 'Work queue depth exceeded' in text:
                        raise WorkQueueFullError
                    text = text.strip() or resp.reason
                    self.logger.error(text)
                    raise DaemonError(text)

    async def _send(self, payload, processor):
        """Send a payload to be converted to JSON.

        Handles temporary connection issues.  Daemon response errors
        are raise through DaemonError.
        """

        def log_error(error):
            nonlocal last_error_log, retry
            now = time.time()
            if now - last_error_log > 60:
                last_error_log = now
                self.logger.error(f'{error}  Retrying occasionally...')
            if retry == self.max_retry and self.failover():
                retry = 0

        on_good_message = None
        last_error_log = 0
        data = json.dumps(payload)
        retry = self.init_retry
        methods = tuple(
            [payload['method']] if isinstance(payload, dict) else [request['method'] for request in payload]
        )
        while True:
            try:
                for method in methods:
                    self.lbrycrd_pending_count_metric.labels(method=method).inc()
                result = await self._send_data(data)
                result = processor(result)
                if on_good_message:
                    self.logger.info(on_good_message)
                return result
            except asyncio.TimeoutError:
                log_error('timeout error.')
            except aiohttp.ServerDisconnectedError:
                log_error('disconnected.')
                on_good_message = 'connection restored'
            except aiohttp.ClientConnectionError:
                log_error('connection problem - is your daemon running?')
                on_good_message = 'connection restored'
            except aiohttp.ClientError as e:
                log_error(f'daemon error: {e}')
                on_good_message = 'running normally'
            except WarmingUpError:
                log_error('starting up checking blocks.')
                on_good_message = 'running normally'
            except WorkQueueFullError:
                log_error('work queue full.')
                on_good_message = 'running normally'
            finally:
                for method in methods:
                    self.lbrycrd_pending_count_metric.labels(method=method).dec()
            await asyncio.sleep(retry)
            retry = max(min(self.max_retry, retry * 2), self.init_retry)

    async def _send_single(self, method, params=None):
        """Send a single request to the daemon."""

        start = time.perf_counter()

        def processor(result):
            err = result['error']
            if not err:
                return result['result']
            if err.get('code') == self.WARMING_UP:
                raise WarmingUpError
            raise DaemonError(err)

        payload = {'method': method, 'id': next(self.id_counter)}
        if params:
            payload['params'] = params
        result = await self._send(payload, processor)
        self.lbrycrd_request_time_metric.labels(method=method).observe(time.perf_counter() - start)
        return result

    async def _send_vector(self, method, params_iterable, replace_errs=False):
        """Send several requests of the same method.

        The result will be an array of the same length as params_iterable.
        If replace_errs is true, any item with an error is returned as None,
        otherwise an exception is raised."""

        start = time.perf_counter()

        def processor(result):
            errs = [item['error'] for item in result if item['error']]
            if any(err.get('code') == self.WARMING_UP for err in errs):
                raise WarmingUpError
            if not errs or replace_errs:
                return [item['result'] for item in result]
            raise DaemonError(errs)

        payload = [{'method': method, 'params': p, 'id': next(self.id_counter)}
                   for p in params_iterable]
        result = []
        if payload:
            result = await self._send(payload, processor)
        self.lbrycrd_request_time_metric.labels(method=method).observe(time.perf_counter() - start)
        return result

    async def _is_rpc_available(self, method):
        """Return whether given RPC method is available in the daemon.

        Results are cached and the daemon will generally not be queried with
        the same method more than once."""
        available = self.available_rpcs.get(method)
        if available is None:
            available = True
            try:
                await self._send_single(method)
            except DaemonError as e:
                err = e.args[0]
                error_code = err.get("code")
                available = error_code != JSONRPC.METHOD_NOT_FOUND
            self.available_rpcs[method] = available
        return available

    async def block_hex_hashes(self, first, count):
        """Return the hex hashes of count block starting at height first."""
        if first + count < (self.cached_height() or 0) - 200:
            return await self._cached_block_hex_hashes(first, count)
        params_iterable = ((h, ) for h in range(first, first + count))
        return await self._send_vector('getblockhash', params_iterable)

    async def _cached_block_hex_hashes(self, first, count):
        """Return the hex hashes of count block starting at height first."""
        cached = self._block_hash_cache.get((first, count))
        if cached:
            return cached
        params_iterable = ((h, ) for h in range(first, first + count))
        self._block_hash_cache[(first, count)] = await self._send_vector('getblockhash', params_iterable)
        return self._block_hash_cache[(first, count)]

    async def deserialised_block(self, hex_hash):
        """Return the deserialised block with the given hex hash."""
        if hex_hash not in self._block_cache:
            block = await self._send_single('getblock', (hex_hash, True))
            self._block_cache[hex_hash] = block
            return block
        return self._block_cache[hex_hash]

    async def raw_blocks(self, hex_hashes):
        """Return the raw binary blocks with the given hex hashes."""
        params_iterable = ((h, False) for h in hex_hashes)
        blocks = await self._send_vector('getblock', params_iterable)
        # Convert hex string to bytes
        return [hex_to_bytes(block) for block in blocks]

    async def mempool_hashes(self):
        """Update our record of the daemon's mempool hashes."""
        return await self._send_single('getrawmempool')

    async def estimatefee(self, block_count):
        """Return the fee estimate for the block count.  Units are whole
        currency units per KB, e.g. 0.00000995, or -1 if no estimate
        is available.
        """
        args = (block_count, )
        if await self._is_rpc_available('estimatesmartfee'):
            estimate = await self._send_single('estimatesmartfee', args)
            return estimate.get('feerate', -1)
        return await self._send_single('estimatefee', args)

    async def getnetworkinfo(self):
        """Return the result of the 'getnetworkinfo' RPC call."""
        return await self._send_single('getnetworkinfo')

    async def relayfee(self):
        """The minimum fee a low-priority tx must pay in order to be accepted
        to the daemon's memory pool."""
        network_info = await self.getnetworkinfo()
        return network_info['relayfee']

    async def getrawtransaction(self, hex_hash, verbose=False):
        """Return the serialized raw transaction with the given hash."""
        # Cast to int because some coin daemons are old and require it
        return await self._send_single('getrawtransaction',
                                       (hex_hash, int(verbose)))

    async def getrawtransactions(self, hex_hashes, replace_errs=True):
        """Return the serialized raw transactions with the given hashes.

        Replaces errors with None by default."""
        params_iterable = ((hex_hash, 0) for hex_hash in hex_hashes)
        txs = await self._send_vector('getrawtransaction', params_iterable,
                                      replace_errs=replace_errs)
        # Convert hex strings to bytes
        return [hex_to_bytes(tx) if tx else None for tx in txs]

    async def broadcast_transaction(self, raw_tx):
        """Broadcast a transaction to the network."""
        return await self._send_single('sendrawtransaction', (raw_tx, ))

    async def height(self):
        """Query the daemon for its current height."""
        self._height = await self._send_single('getblockcount')
        return self._height

    def cached_height(self):
        """Return the cached daemon height.

        If the daemon has not been queried yet this returns None."""
        return self._height


def handles_errors(decorated_function):
    @wraps(decorated_function)
    async def wrapper(*args, **kwargs):
        try:
            return await decorated_function(*args, **kwargs)
        except DaemonError as daemon_error:
            raise RPCError(1, daemon_error.args[0])
    return wrapper


class LBCDaemon(Daemon):
    @handles_errors
    async def getrawtransaction(self, hex_hash, verbose=False):
        return await super().getrawtransaction(hex_hash=hex_hash, verbose=verbose)

    @handles_errors
    async def getclaimbyid(self, claim_id):
        '''Given a claim id, retrieves claim information.'''
        return await self._send_single('getclaimbyid', (claim_id,))

    @handles_errors
    async def getclaimsbyids(self, claim_ids):
        '''Given a list of claim ids, batches calls to retrieve claim information.'''
        return await self._send_vector('getclaimbyid', ((claim_id,) for claim_id in claim_ids))

    @handles_errors
    async def getclaimsforname(self, name):
        '''Given a name, retrieves all claims matching that name.'''
        return await self._send_single('getclaimsforname', (name,))

    @handles_errors
    async def getclaimsfortx(self, txid):
        '''Given a txid, returns the claims it make.'''
        return await self._send_single('getclaimsfortx', (txid,)) or []

    @handles_errors
    async def getnameproof(self, name, block_hash=None):
        '''Given a name and optional block_hash, returns a name proof and winner, if any.'''
        return await self._send_single('getnameproof', (name, block_hash,) if block_hash else (name,))

    @handles_errors
    async def getvalueforname(self, name):
        '''Given a name, returns the winning claim value.'''
        return await self._send_single('getvalueforname', (name,))

    @handles_errors
    async def claimname(self, name, hexvalue, amount):
        '''Claim a name, used for functional tests only.'''
        return await self._send_single('claimname', (name, hexvalue, float(amount)))

1	import asyncio	5✔
2	import itertools	5✔
3	import json	5✔
4	import time	5✔
5	from functools import wraps	5✔
6
7	import aiohttp	5✔
8	from prometheus_client import Gauge, Histogram	5✔
9	from lbry.utils import LRUCacheWithMetrics	5✔
10	from lbry.wallet.rpc.jsonrpc import RPCError	5✔
11	from lbry.wallet.server.util import hex_to_bytes, class_logger	5✔
12	from lbry.wallet.rpc import JSONRPC	5✔
13
14
15	class DaemonError(Exception):	5✔
16	"""Raised when the daemon returns an error in its results."""
17
18
19	class WarmingUpError(Exception):	5✔
20	"""Internal - when the daemon is warming up."""
21
22
23	class WorkQueueFullError(Exception):	5✔
24	"""Internal - when the daemon's work queue is full."""
25
26
27	NAMESPACE = "wallet_server"	5✔
28
29
30	class Daemon:	5✔
31	"""Handles connections to a daemon at the given URL."""
32
33	WARMING_UP = -28	5✔
34	id_counter = itertools.count()	5✔
35
36	lbrycrd_request_time_metric = Histogram(	5✔
37	"lbrycrd_request", "lbrycrd requests count", namespace=NAMESPACE, labelnames=("method",)
38	)
39	lbrycrd_pending_count_metric = Gauge(	5✔
40	"lbrycrd_pending_count", "Number of lbrycrd rpcs that are in flight", namespace=NAMESPACE,
41	labelnames=("method",)
42	)
43
44	def __init__(self, coin, url, max_workqueue=10, init_retry=0.25,	5✔
45	max_retry=4.0):
46	self.coin = coin	4✔
47	self.logger = class_logger(__name__, self.__class__.__name__)	4✔
48	self.set_url(url)	4✔
49	# Limit concurrent RPC calls to this number.
50	# See DEFAULT_HTTP_WORKQUEUE in bitcoind, which is typically 16
51	self.workqueue_semaphore = asyncio.Semaphore(value=max_workqueue)	4✔
52	self.init_retry = init_retry	4✔
53	self.max_retry = max_retry	4✔
54	self._height = None	4✔
55	self.available_rpcs = {}	4✔
56	self.connector = aiohttp.TCPConnector()	4✔
57	self._block_hash_cache = LRUCacheWithMetrics(100000)	4✔
58	self._block_cache = LRUCacheWithMetrics(2 ** 16, metric_name='block', namespace=NAMESPACE)	4✔
59
60	async def close(self):	5✔
61	if self.connector:	4!
62	await self.connector.close()	4✔
63	self.connector = None	4✔
64
65	def set_url(self, url):	5✔
66	"""Set the URLS to the given list, and switch to the first one."""
67	urls = url.split(',')	4✔
68	urls = [self.coin.sanitize_url(url) for url in urls]	4!
69	for n, url in enumerate(urls):	4!
70	status = '' if n else ' (current)'	4✔
71	logged_url = self.logged_url(url)	4✔
72	self.logger.info(f'daemon #{n + 1} at {logged_url}{status}')	4✔
73	self.url_index = 0	4✔
74	self.urls = urls	4✔
75
76	def current_url(self):	5✔
77	"""Returns the current daemon URL."""
78	return self.urls[self.url_index]	4✔
79
80	def logged_url(self, url=None):	5✔
81	"""The host and port part, for logging."""
82	url = url or self.current_url()	4✔
83	return url[url.rindex('@') + 1:]	4✔
84
85	def failover(self):	5✔
86	"""Call to fail-over to the next daemon URL.
87
88	Returns False if there is only one, otherwise True.
89	"""
90	if len(self.urls) > 1:	×
91	self.url_index = (self.url_index + 1) % len(self.urls)	×
92	self.logger.info(f'failing over to {self.logged_url()}')	×
93	return True	×
94	return False	×
95
96	def client_session(self):	5✔
97	"""An aiohttp client session."""
98	return aiohttp.ClientSession(connector=self.connector, connector_owner=False)	4✔
99
100	async def _send_data(self, data):	5✔
101	if not self.connector:	4!
102	raise asyncio.CancelledError('Tried to send request during shutdown.')	4✔
103	async with self.workqueue_semaphore:	4✔
104	async with self.client_session() as session:	4✔
105	async with session.post(self.current_url(), data=data) as resp:	4✔
106	kind = resp.headers.get('Content-Type', None)	4✔
107	if kind == 'application/json':	4!
108	return await resp.json()	4✔
109	# bitcoind's HTTP protocol "handling" is a bad joke
110	text = await resp.text()	×
111	if 'Work queue depth exceeded' in text:	×
112	raise WorkQueueFullError	×
113	text = text.strip() or resp.reason	×
114	self.logger.error(text)	×
115	raise DaemonError(text)	×
116
117	async def _send(self, payload, processor):	5✔
118	"""Send a payload to be converted to JSON.
119
120	Handles temporary connection issues. Daemon response errors
121	are raise through DaemonError.
122	"""
123
124	def log_error(error):	4✔
125	nonlocal last_error_log, retry
126	now = time.time()	2✔
127	if now - last_error_log > 60:	2!
128	last_error_log = now	2✔
129	self.logger.error(f'{error} Retrying occasionally...')	2✔
130	if retry == self.max_retry and self.failover():	2!
131	retry = 0	×
132
133	on_good_message = None	4✔
134	last_error_log = 0	4✔
135	data = json.dumps(payload)	4✔
136	retry = self.init_retry	4✔
137	methods = tuple(	4!
138	[payload['method']] if isinstance(payload, dict) else [request['method'] for request in payload]
139	)
140	while True:	4✔
141	try:	4✔
142	for method in methods:	4!
143	self.lbrycrd_pending_count_metric.labels(method=method).inc()	4✔
144	result = await self._send_data(data)	4✔
145	result = processor(result)	4✔
146	if on_good_message:	4!
147	self.logger.info(on_good_message)	2✔
148	return result	4✔
149	except asyncio.TimeoutError:	4!
150	log_error('timeout error.')	×
151	except aiohttp.ServerDisconnectedError:	4!
UNCOV 152	log_error('disconnected.')	×
UNCOV 153	on_good_message = 'connection restored'	×
154	except aiohttp.ClientConnectionError:	4!
155	log_error('connection problem - is your daemon running?')	2✔
156	on_good_message = 'connection restored'	2✔
157	except aiohttp.ClientError as e:	4!
158	log_error(f'daemon error: {e}')	×
159	on_good_message = 'running normally'	×
160	except WarmingUpError:	4!
161	log_error('starting up checking blocks.')	1✔
162	on_good_message = 'running normally'	1✔
163	except WorkQueueFullError:	4✔
164	log_error('work queue full.')	×
165	on_good_message = 'running normally'	×
166	finally:
167	for method in methods:	4!
168	self.lbrycrd_pending_count_metric.labels(method=method).dec()	4✔
169	await asyncio.sleep(retry)	2✔
170	retry = max(min(self.max_retry, retry * 2), self.init_retry)	2✔
171
172	async def _send_single(self, method, params=None):	5✔
173	"""Send a single request to the daemon."""
174
175	start = time.perf_counter()	4✔
176
177	def processor(result):	4✔
178	err = result['error']	4✔
179	if not err:	4!
180	return result['result']	4✔
181	if err.get('code') == self.WARMING_UP:	2!
182	raise WarmingUpError	1✔
183	raise DaemonError(err)	2✔
184
185	payload = {'method': method, 'id': next(self.id_counter)}	4✔
186	if params:	4!
187	payload['params'] = params	4✔
188	result = await self._send(payload, processor)	4✔
189	self.lbrycrd_request_time_metric.labels(method=method).observe(time.perf_counter() - start)	4✔
190	return result	4✔
191
192	async def _send_vector(self, method, params_iterable, replace_errs=False):	5✔
193	"""Send several requests of the same method.
194
195	The result will be an array of the same length as params_iterable.
196	If replace_errs is true, any item with an error is returned as None,
197	otherwise an exception is raised."""
198
199	start = time.perf_counter()	4✔
200
201	def processor(result):	4✔
202	errs = [item['error'] for item in result if item['error']]	4!
203	if any(err.get('code') == self.WARMING_UP for err in errs):	4!
204	raise WarmingUpError	×
205	if not errs or replace_errs:	4!
206	return [item['result'] for item in result]	4!
207	raise DaemonError(errs)	×
208
209	payload = [{'method': method, 'params': p, 'id': next(self.id_counter)}	4!
210	for p in params_iterable]
211	result = []	4✔
212	if payload:	4!
213	result = await self._send(payload, processor)	4✔
214	self.lbrycrd_request_time_metric.labels(method=method).observe(time.perf_counter() - start)	4✔
215	return result	4✔
216
217	async def _is_rpc_available(self, method):	5✔
218	"""Return whether given RPC method is available in the daemon.
219
220	Results are cached and the daemon will generally not be queried with
221	the same method more than once."""
222	available = self.available_rpcs.get(method)	×
223	if available is None:	×
224	available = True	×
225	try:	×
226	await self._send_single(method)	×
227	except DaemonError as e:	×
228	err = e.args[0]	×
229	error_code = err.get("code")	×
230	available = error_code != JSONRPC.METHOD_NOT_FOUND	×
231	self.available_rpcs[method] = available	×
232	return available	×
233
234	async def block_hex_hashes(self, first, count):	5✔
235	"""Return the hex hashes of count block starting at height first."""
236	if first + count < (self.cached_height() or 0) - 200:	4!
237	return await self._cached_block_hex_hashes(first, count)	2✔
238	params_iterable = ((h, ) for h in range(first, first + count))	4!
239	return await self._send_vector('getblockhash', params_iterable)	4✔
240
241	async def _cached_block_hex_hashes(self, first, count):	5✔
242	"""Return the hex hashes of count block starting at height first."""
243	cached = self._block_hash_cache.get((first, count))	2✔
244	if cached:	2!
245	return cached	×
246	params_iterable = ((h, ) for h in range(first, first + count))	2!
247	self._block_hash_cache[(first, count)] = await self._send_vector('getblockhash', params_iterable)	2✔
248	return self._block_hash_cache[(first, count)]	2✔
249
250	async def deserialised_block(self, hex_hash):	5✔
251	"""Return the deserialised block with the given hex hash."""
252	if hex_hash not in self._block_cache:	4!
253	block = await self._send_single('getblock', (hex_hash, True))	4✔
254	self._block_cache[hex_hash] = block	4✔
255	return block	4✔
256	return self._block_cache[hex_hash]	4✔
257
258	async def raw_blocks(self, hex_hashes):	5✔
259	"""Return the raw binary blocks with the given hex hashes."""
260	params_iterable = ((h, False) for h in hex_hashes)	4!
261	blocks = await self._send_vector('getblock', params_iterable)	4✔
262	# Convert hex string to bytes
263	return [hex_to_bytes(block) for block in blocks]	4!
264
265	async def mempool_hashes(self):	5✔
266	"""Update our record of the daemon's mempool hashes."""
267	return await self._send_single('getrawmempool')	4✔
268
269	async def estimatefee(self, block_count):	5✔
270	"""Return the fee estimate for the block count. Units are whole
271	currency units per KB, e.g. 0.00000995, or -1 if no estimate
272	is available.
273	"""
274	args = (block_count, )	×
275	if await self._is_rpc_available('estimatesmartfee'):	×
276	estimate = await self._send_single('estimatesmartfee', args)	×
277	return estimate.get('feerate', -1)	×
278	return await self._send_single('estimatefee', args)	×
279
280	async def getnetworkinfo(self):	5✔
281	"""Return the result of the 'getnetworkinfo' RPC call."""
282	return await self._send_single('getnetworkinfo')	×
283
284	async def relayfee(self):	5✔
285	"""The minimum fee a low-priority tx must pay in order to be accepted
286	to the daemon's memory pool."""
287	network_info = await self.getnetworkinfo()	×
288	return network_info['relayfee']	×
289
290	async def getrawtransaction(self, hex_hash, verbose=False):	5✔
291	"""Return the serialized raw transaction with the given hash."""
292	# Cast to int because some coin daemons are old and require it
293	return await self._send_single('getrawtransaction',	4✔
294	(hex_hash, int(verbose)))
295
296	async def getrawtransactions(self, hex_hashes, replace_errs=True):	5✔
297	"""Return the serialized raw transactions with the given hashes.
298
299	Replaces errors with None by default."""
300	params_iterable = ((hex_hash, 0) for hex_hash in hex_hashes)	4!
301	txs = await self._send_vector('getrawtransaction', params_iterable,	4✔
302	replace_errs=replace_errs)
303	# Convert hex strings to bytes
304	return [hex_to_bytes(tx) if tx else None for tx in txs]	4!
305
306	async def broadcast_transaction(self, raw_tx):	5✔
307	"""Broadcast a transaction to the network."""
308	return await self._send_single('sendrawtransaction', (raw_tx, ))	4✔
309
310	async def height(self):	5✔
311	"""Query the daemon for its current height."""
312	self._height = await self._send_single('getblockcount')	4✔
313	return self._height	4✔
314
315	def cached_height(self):	5✔
316	"""Return the cached daemon height.
317
318	If the daemon has not been queried yet this returns None."""
319	return self._height	4✔
320
321
322	def handles_errors(decorated_function):	5✔
323	@wraps(decorated_function)	5✔
324	async def wrapper(args, *kwargs):
325	try:	4✔
326	return await decorated_function(args, *kwargs)	4✔
327	except DaemonError as daemon_error:	3✔
328	raise RPCError(1, daemon_error.args[0])	2✔
329	return wrapper	5✔
330
331
332	class LBCDaemon(Daemon):	5✔
333	@handles_errors	5✔
334	async def getrawtransaction(self, hex_hash, verbose=False):	5✔
335	return await super().getrawtransaction(hex_hash=hex_hash, verbose=verbose)	4✔
336
337	@handles_errors	5✔
338	async def getclaimbyid(self, claim_id):
339	'''Given a claim id, retrieves claim information.'''
340	return await self._send_single('getclaimbyid', (claim_id,))	×
341
342	@handles_errors	5✔
343	async def getclaimsbyids(self, claim_ids):
344	'''Given a list of claim ids, batches calls to retrieve claim information.'''
345	return await self._send_vector('getclaimbyid', ((claim_id,) for claim_id in claim_ids))	×
346
347	@handles_errors	5✔
348	async def getclaimsforname(self, name):
349	'''Given a name, retrieves all claims matching that name.'''
350	return await self._send_single('getclaimsforname', (name,))	×
351
352	@handles_errors	5✔
353	async def getclaimsfortx(self, txid):
354	'''Given a txid, returns the claims it make.'''
355	return await self._send_single('getclaimsfortx', (txid,)) or []	×
356
357	@handles_errors	5✔
358	async def getnameproof(self, name, block_hash=None):	5✔
359	'''Given a name and optional block_hash, returns a name proof and winner, if any.'''
360	return await self._send_single('getnameproof', (name, block_hash,) if block_hash else (name,))	×
361
362	@handles_errors	5✔
363	async def getvalueforname(self, name):
364	'''Given a name, returns the winning claim value.'''
365	return await self._send_single('getvalueforname', (name,))	×
366
367	@handles_errors	5✔
368	async def claimname(self, name, hexvalue, amount):
369	'''Claim a name, used for functional tests only.'''
370	return await self._send_single('claimname', (name, hexvalue, float(amount)))	×

lbryio / lbry-sdk / 1263745490

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous