19944567212

Committed 04 Dec 2025 09:29PM UTC coverage: 90.921% (-0.02%) from 90.944%

Build # 19944567212

Build Type

push

github

Committed by Cryp Toon

Commit Message

RELEASE 0.7.6 - MESSAGE SIGNING AND DETERMINISTIC SIGNATURES

Coverage Stats

8042 of 8845 relevant lines covered (90.92%)

1.81 hits per line

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

86.62

/bitcoinlib/services/blockcypher.py

# -*- coding: utf-8 -*-
#
#    BitcoinLib - Python Cryptocurrency Library
#    BlockCypher client
#    © 2017-2023 May - 1200 Web Development <http://1200wd.com/>
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU Affero General Public License as
#    published by the Free Software Foundation, either version 3 of the
#    License, or (at your option) any later version.
#
#    This program is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    GNU Affero General Public License for more details.
#
#    You should have received a copy of the GNU Affero General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import logging
from datetime import datetime
from datetime import timezone
from bitcoinlib.main import MAX_TRANSACTIONS
from bitcoinlib.services.baseclient import BaseClient, ClientError
from bitcoinlib.transactions import Transaction

PROVIDERNAME = 'blockcypher'

_logger = logging.getLogger(__name__)


class BlockCypher(BaseClient):

    def __init__(self, network, base_url, denominator, *args):
        super(self.__class__, self).__init__(network, PROVIDERNAME, base_url, denominator, *args)

    def compose_request(self, function, data, parameter='', variables=None, method='get'):
        url_path = function + '/' + data
        if parameter:
            url_path += '/' + parameter
        if variables is None:
            variables = {}
        if self.api_key:
            variables.update({'token': self.api_key})
        return self.request(url_path, variables, method)

    def getbalance(self, addresslist):
        addresslist = self._addresslist_convert(addresslist)
        addresses = ';'.join([a.address for a in addresslist])
        res = self.compose_request('addrs', addresses, 'balance')
        balance = 0.0
        if not isinstance(res, list):
            res = [res]
        for rec in res:
            balance += float(rec['final_balance'])
        return round(balance * self.units)

    def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS):
        address = self._address_convert(address)
        res = self.compose_request('addrs', address.address, variables={'unspentOnly': 1, 'limit': 2000})
        transactions = []
        if not isinstance(res, list):
            res = [res]
        for a in res:
            txrefs = a.setdefault('txrefs', []) + a.get('unconfirmed_txrefs', [])
            if len(txrefs) > 500:
                _logger.warning("BlockCypher: Large number of transactions for address %s, "
                                "Transaction list may be incomplete" % address)
            for tx in txrefs:
                if tx['tx_hash'] == after_txid:
                    break
                tdate = None
                if 'confirmed' in tx:
                    try:
                        tdate = datetime.strptime(tx['confirmed'], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
                    except ValueError:
                        tdate = datetime.strptime(tx['confirmed'], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc)
                transactions.append({
                    'address': address.address_orig,
                    'txid': tx['tx_hash'],
                    'confirmations': tx['confirmations'],
                    'output_n': tx['tx_output_n'],
                    'index': 0,
                    'value': round(tx['value'] * self.units),
                    'script': '',
                    'block_height': None,
                    'date': tdate
                })
        return transactions[::-1][:limit]

    def gettransaction(self, txid):
        tx = self.compose_request('txs', txid, variables={'includeHex': 'true'})
        t = Transaction.parse_hex(tx['hex'], strict=self.strict, network=self.network)
        if tx['confirmations']:
            t.status = 'confirmed'
            t.date = datetime.strptime(tx['confirmed'][:19], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc)
        else:
            t.status = 'unconfirmed'
        t.confirmations = tx['confirmations']
        t.block_height = tx['block_height'] if tx['block_height'] > 0 else None
        t.fee = tx['fees']
        t.rawtx = bytes.fromhex(tx['hex'])
        t.size = round(len(tx['hex']) / 2)
        t.network = self.network
        t.input_total = 0
        if len(t.inputs) != len(tx['inputs']):
            raise ClientError("Invalid number of inputs provided. Raw tx: %d, blockcypher: %d" %
                              (len(t.inputs), len(tx['inputs'])))
        for n, i in enumerate(t.inputs):
            if not t.coinbase and not (tx['inputs'][n]['output_index'] == i.output_n_int and
                                       tx['inputs'][n]['prev_hash'] == i.prev_txid.hex()):
                raise ClientError("Transaction inputs do not match raw transaction")
            if 'output_value' in tx['inputs'][n]:
                if not t.coinbase:
                    i.value = tx['inputs'][n]['output_value']
                t.input_total += i.value
        if len(t.outputs) != len(tx['outputs']):
            raise ClientError("Invalid number of outputs provided. Raw tx: %d, blockcypher: %d" %
                              (len(t.outputs), len(tx['outputs'])))
        for n, o in enumerate(t.outputs):
            if 'spent_by' in tx['outputs'][n]:
                o.spent = True
                o.spending_txid = tx['outputs'][n]['spent_by']
        return t

    def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS):
        txs = []
        address = self._address_convert(address)
        res = self.compose_request('addrs', address.address, variables={'unspentOnly': 0, 'limit': 2000})
        if not isinstance(res, list):
            res = [res]
        for a in res:
            txrefs = a.get('txrefs', []) + a.get('unconfirmed_txrefs', [])
            txids = []
            for t in txrefs[::-1]:
                if t['tx_hash'] not in txids:
                    txids.append(t['tx_hash'])
                if t['tx_hash'] == after_txid:
                    txids = []
            if len(txids) > 500:
                _logger.info("BlockCypher: Large number of transactions for address %s, "
                             "Transaction list may be incomplete" % address.address_orig)
            for txid in txids[:limit]:
                t = self.gettransaction(txid)
                txs.append(t)
        return txs

    def getrawtransaction(self, txid):
        return self.compose_request('txs', txid, variables={'includeHex': 'true'})['hex']

    def sendrawtransaction(self, rawtx):
        # BlockCypher sometimes accepts transactions, but does not push them to the network :(
        if self.network.name in ['bitcoin', 'litecoin']:
            raise ClientError("Avoid stuck transactions, skip usage of blockcypher provider")
        res = self.compose_request('txs', 'push', variables={'tx': rawtx}, method='post')
        return {
            'txid': res['tx']['hash'],
            'response_dict': res
        }

    def estimatefee(self, blocks):
        res = self.compose_request('', '')
        if blocks <= 10:
            return res['medium_fee_per_kb']
        else:
            return res['low_fee_per_kb']

    def blockcount(self):
        return self.compose_request('', '')['height']

    def mempool(self, txid):
        if txid:
            tx = self.compose_request('txs', txid)
            if tx['confirmations'] == 0:
                return [tx['hash']]
            else:
                return []
        return False

    def getblock(self, blockid, parse_transactions, page, limit):
        if limit > 100:
            limit = 100
        bd = self.compose_request('blocks', str(blockid), variables={'limit': limit, 'txstart': ((page-1)*limit)})
        if parse_transactions:
            txs = []
            for txid in bd['txids']:
                try:
                    txs.append(self.gettransaction(txid))
                except Exception as e:
                    _logger.error("Could not parse tx %s with error %s" % (txid, e))
        else:
            txs = bd['txids']

        block = {
            'bits': bd['bits'],
            'depth': bd['depth'],
            'block_hash': bd['hash'],
            'height': bd['height'],
            'merkle_root': bd['mrkl_root'],
            'nonce': bd['nonce'],
            'prev_block': bd['prev_block'],
            'time': int(datetime.strptime(bd['time'], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc).timestamp()),
            'tx_count': bd['n_tx'],
            'txs': txs,
            'version': bd['ver'],
            'page': page,
            'pages': None if not limit else int(bd['n_tx'] // limit) + (bd['n_tx'] % limit > 0),
            'limit': limit
        }
        return block

    # def getrawblock(self, blockid):

    def isspent(self, txid, output_n):
        t = self.gettransaction(txid)
        return 1 if t.outputs[output_n].spent else 0

    def getinfo(self):
        info = self.compose_request('', '')
        return {
            'blockcount': info['height'],
            'chain': info['name'],
            'difficulty': 0,
            'hashrate': 0,
            'mempool_size': info['unconfirmed_count'],
        }

1	# -- coding: utf-8 --
2	#
3	# BitcoinLib - Python Cryptocurrency Library
4	# BlockCypher client
5	# © 2017-2023 May - 1200 Web Development <http://1200wd.com/>
6	#
7	# This program is free software: you can redistribute it and/or modify
8	# it under the terms of the GNU Affero General Public License as
9	# published by the Free Software Foundation, either version 3 of the
10	# License, or (at your option) any later version.
11	#
12	# This program is distributed in the hope that it will be useful,
13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15	# GNU Affero General Public License for more details.
16	#
17	# You should have received a copy of the GNU Affero General Public License
18	# along with this program. If not, see <http://www.gnu.org/licenses/>.
19	#
20
21	import logging	2✔
22	from datetime import datetime	2✔
23	from datetime import timezone	2✔
24	from bitcoinlib.main import MAX_TRANSACTIONS	2✔
25	from bitcoinlib.services.baseclient import BaseClient, ClientError	2✔
26	from bitcoinlib.transactions import Transaction	2✔
27
28	PROVIDERNAME = 'blockcypher'	2✔
29
30	_logger = logging.getLogger(__name__)	2✔
31
32
33	class BlockCypher(BaseClient):	2✔
34
35	def __init__(self, network, base_url, denominator, *args):	2✔
36	super(self.__class__, self).__init__(network, PROVIDERNAME, base_url, denominator, *args)	2✔
37
38	def compose_request(self, function, data, parameter='', variables=None, method='get'):	2✔
39	url_path = function + '/' + data	2✔
40	if parameter:	2✔
41	url_path += '/' + parameter	2✔
42	if variables is None:	2✔
43	variables = {}	2✔
44	if self.api_key:	2✔
45	variables.update({'token': self.api_key})	×
46	return self.request(url_path, variables, method)	2✔
47
48	def getbalance(self, addresslist):	2✔
49	addresslist = self._addresslist_convert(addresslist)	2✔
50	addresses = ';'.join([a.address for a in addresslist])	2✔
51	res = self.compose_request('addrs', addresses, 'balance')	2✔
52	balance = 0.0	2✔
53	if not isinstance(res, list):	2✔
54	res = [res]	2✔
55	for rec in res:	2✔
56	balance += float(rec['final_balance'])	2✔
57	return round(balance * self.units)	2✔
58
59	def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS):	2✔
60	address = self._address_convert(address)	2✔
61	res = self.compose_request('addrs', address.address, variables={'unspentOnly': 1, 'limit': 2000})	2✔
62	transactions = []	2✔
63	if not isinstance(res, list):	2✔
64	res = [res]	2✔
65	for a in res:	2✔
66	txrefs = a.setdefault('txrefs', []) + a.get('unconfirmed_txrefs', [])	2✔
67	if len(txrefs) > 500:	2✔
68	_logger.warning("BlockCypher: Large number of transactions for address %s, "	×
69	"Transaction list may be incomplete" % address)
70	for tx in txrefs:	2✔
71	if tx['tx_hash'] == after_txid:	2✔
72	break	2✔
73	tdate = None	2✔
74	if 'confirmed' in tx:	2✔
75	try:	2✔
76	tdate = datetime.strptime(tx['confirmed'], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)	2✔
77	except ValueError:	×
78	tdate = datetime.strptime(tx['confirmed'], "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=timezone.utc)	×
79	transactions.append({	2✔
80	'address': address.address_orig,
81	'txid': tx['tx_hash'],
82	'confirmations': tx['confirmations'],
83	'output_n': tx['tx_output_n'],
84	'index': 0,
85	'value': round(tx['value'] * self.units),
86	'script': '',
87	'block_height': None,
88	'date': tdate
89	})
90	return transactions[::-1][:limit]	2✔
91
92	def gettransaction(self, txid):	2✔
93	tx = self.compose_request('txs', txid, variables={'includeHex': 'true'})	2✔
94	t = Transaction.parse_hex(tx['hex'], strict=self.strict, network=self.network)	2✔
95	if tx['confirmations']:	2✔
96	t.status = 'confirmed'	2✔
97	t.date = datetime.strptime(tx['confirmed'][:19], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc)	2✔
98	else:
99	t.status = 'unconfirmed'	×
100	t.confirmations = tx['confirmations']	2✔
101	t.block_height = tx['block_height'] if tx['block_height'] > 0 else None	2✔
102	t.fee = tx['fees']	2✔
103	t.rawtx = bytes.fromhex(tx['hex'])	2✔
104	t.size = round(len(tx['hex']) / 2)	2✔
105	t.network = self.network	2✔
106	t.input_total = 0	2✔
107	if len(t.inputs) != len(tx['inputs']):	2✔
108	raise ClientError("Invalid number of inputs provided. Raw tx: %d, blockcypher: %d" %	1✔
109	(len(t.inputs), len(tx['inputs'])))
110	for n, i in enumerate(t.inputs):	2✔
111	if not t.coinbase and not (tx['inputs'][n]['output_index'] == i.output_n_int and	2✔
112	tx['inputs'][n]['prev_hash'] == i.prev_txid.hex()):
113	raise ClientError("Transaction inputs do not match raw transaction")	×
114	if 'output_value' in tx['inputs'][n]:	2✔
115	if not t.coinbase:	2✔
116	i.value = tx['inputs'][n]['output_value']	2✔
117	t.input_total += i.value	2✔
118	if len(t.outputs) != len(tx['outputs']):	2✔
119	raise ClientError("Invalid number of outputs provided. Raw tx: %d, blockcypher: %d" %	×
120	(len(t.outputs), len(tx['outputs'])))
121	for n, o in enumerate(t.outputs):	2✔
122	if 'spent_by' in tx['outputs'][n]:	2✔
123	o.spent = True	2✔
124	o.spending_txid = tx['outputs'][n]['spent_by']	2✔
125	return t	2✔
126
127	def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS):	2✔
128	txs = []	2✔
129	address = self._address_convert(address)	2✔
130	res = self.compose_request('addrs', address.address, variables={'unspentOnly': 0, 'limit': 2000})	2✔
131	if not isinstance(res, list):	2✔
132	res = [res]	2✔
133	for a in res:	2✔
134	txrefs = a.get('txrefs', []) + a.get('unconfirmed_txrefs', [])	2✔
135	txids = []	2✔
136	for t in txrefs[::-1]:	2✔
137	if t['tx_hash'] not in txids:	2✔
138	txids.append(t['tx_hash'])	2✔
139	if t['tx_hash'] == after_txid:	2✔
140	txids = []	2✔
141	if len(txids) > 500:	2✔
142	_logger.info("BlockCypher: Large number of transactions for address %s, "	×
143	"Transaction list may be incomplete" % address.address_orig)
144	for txid in txids[:limit]:	2✔
145	t = self.gettransaction(txid)	2✔
146	txs.append(t)	2✔
147	return txs	2✔
148
149	def getrawtransaction(self, txid):	2✔
150	return self.compose_request('txs', txid, variables={'includeHex': 'true'})['hex']	1✔
151
152	def sendrawtransaction(self, rawtx):	2✔
153	# BlockCypher sometimes accepts transactions, but does not push them to the network :(
154	if self.network.name in ['bitcoin', 'litecoin']:	×
155	raise ClientError("Avoid stuck transactions, skip usage of blockcypher provider")	×
156	res = self.compose_request('txs', 'push', variables={'tx': rawtx}, method='post')	×
157	return {	×
158	'txid': res['tx']['hash'],
159	'response_dict': res
160	}
161
162	def estimatefee(self, blocks):	2✔
163	res = self.compose_request('', '')	2✔
164	if blocks <= 10:	2✔
165	return res['medium_fee_per_kb']	2✔
166	else:
167	return res['low_fee_per_kb']	×
168
169	def blockcount(self):	2✔
170	return self.compose_request('', '')['height']	2✔
171
172	def mempool(self, txid):	2✔
173	if txid:	2✔
174	tx = self.compose_request('txs', txid)	2✔
175	if tx['confirmations'] == 0:	2✔
176	return [tx['hash']]	×
177	else:
178	return []	2✔
179	return False	×
180
181	def getblock(self, blockid, parse_transactions, page, limit):	2✔
182	if limit > 100:	2✔
183	limit = 100	×
184	bd = self.compose_request('blocks', str(blockid), variables={'limit': limit, 'txstart': ((page-1)*limit)})	2✔
185	if parse_transactions:	2✔
186	txs = []	2✔
187	for txid in bd['txids']:	2✔
188	try:	2✔
189	txs.append(self.gettransaction(txid))	2✔
190	except Exception as e:	×
191	_logger.error("Could not parse tx %s with error %s" % (txid, e))	×
192	else:
193	txs = bd['txids']	×
194
195	block = {	2✔
196	'bits': bd['bits'],
197	'depth': bd['depth'],
198	'block_hash': bd['hash'],
199	'height': bd['height'],
200	'merkle_root': bd['mrkl_root'],
201	'nonce': bd['nonce'],
202	'prev_block': bd['prev_block'],
203	'time': int(datetime.strptime(bd['time'], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc).timestamp()),
204	'tx_count': bd['n_tx'],
205	'txs': txs,
206	'version': bd['ver'],
207	'page': page,
208	'pages': None if not limit else int(bd['n_tx'] // limit) + (bd['n_tx'] % limit > 0),
209	'limit': limit
210	}
211	return block	2✔
212
213	# def getrawblock(self, blockid):
214
215	def isspent(self, txid, output_n):	2✔
216	t = self.gettransaction(txid)	1✔
217	return 1 if t.outputs[output_n].spent else 0	1✔
218
219	def getinfo(self):	2✔
220	info = self.compose_request('', '')	1✔
221	return {	1✔
222	'blockcount': info['height'],
223	'chain': info['name'],
224	'difficulty': 0,
225	'hashrate': 0,
226	'mempool_size': info['unconfirmed_count'],
227	}

1200wd / bitcoinlib / 19944567212

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