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

93.48

/bitcoinlib/services/mempool.py

# -*- coding: utf-8 -*-
#
#    BitcoinLib - Python Cryptocurrency Library
#    mempool.space client
#    © 2021-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, timezone
from bitcoinlib.main import MAX_TRANSACTIONS
from bitcoinlib.services.baseclient import BaseClient, ClientError
from bitcoinlib.transactions import Transaction
from bitcoinlib.encoding import varstr

PROVIDERNAME = 'mempool'
# Please note: In the Blockstream API, the first couple of Bitcoin blocks are not correctly indexed,
# so transactions from these blocks are missing.

_logger = logging.getLogger(__name__)


class MempoolClient(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='', parameter2='', variables=None, post_data='',
                        method='get'):
        url_path = function
        if data:
            url_path += '/' + data
        if parameter:
            url_path += '/' + parameter
        if parameter2:
            url_path += '/' + parameter2
        if variables is None:
            variables = {}
        if self.api_key:
            variables.update({'token': self.api_key})
        return self.request(url_path, variables, method, post_data=post_data)

    def getbalance(self, addresslist):
        balance = 0
        for address in addresslist:
            res = self.compose_request('address', address)
            balance += res['chain_stats']['funded_txo_sum'] - res['chain_stats']['spent_txo_sum']
        return balance

    def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS):
        self.latest_block = self.blockcount() if not self.latest_block else self.latest_block
        res = self.compose_request('address', address, 'utxo')
        utxos = []
        # # key=lambda k: (k[2], pow(10, 20)-k[0].transaction_id, k[3]), reverse=True
        res = sorted(res, key=lambda k: 0 if 'block_height' not in k['status'] else k['status']['block_height'])
        for a in res:
            confirmations = 0
            block_height = None
            if 'block_height' in a['status']:
                block_height = a['status']['block_height']
                confirmations = self.latest_block - block_height
            utxos.append({
                'address': address,
                'txid': a['txid'],
                'confirmations': confirmations,
                'output_n': a['vout'],
                'input_n': 0,
                'block_height': block_height,
                'fee': None,
                'size': 0,
                'value': a['value'],
                'script': '',
                'date': None if 'block_time' not in a['status'] else
                datetime.fromtimestamp(a['status']['block_time'], timezone.utc)
            })
            if a['txid'] == after_txid:
                utxos = []
        return utxos[:limit]

    def _parse_transaction(self, tx):
        block_height = None if 'block_height' not in tx['status'] else tx['status']['block_height']
        confirmations = 0
        tx_date = None
        status = 'unconfirmed'
        if tx['status']['confirmed']:
            if block_height:
                self.latest_block = self.blockcount() if not self.latest_block else self.latest_block
                confirmations = self.latest_block - block_height + 1
            tx_date = datetime.fromtimestamp(tx['status']['block_time'], timezone.utc)
            status = 'confirmed'
        witness_type = 'legacy'
        if tx['size'] * 4 > tx['weight']:
            witness_type = 'segwit'

        t = Transaction(locktime=tx['locktime'], version=tx['version'], network=self.network, block_height=block_height,
                        fee=tx['fee'], size=tx['size'], txid=tx['txid'], date=tx_date, confirmations=confirmations,
                        status=status, coinbase=tx['vin'][0]['is_coinbase'], witness_type=witness_type)
        for ti in tx['vin']:
            if ti['is_coinbase']:
                t.add_input(prev_txid=ti['txid'], output_n=ti['vout'], unlocking_script=ti['scriptsig'], value=0,
                            sequence=ti['sequence'], strict=self.strict, witness_type=witness_type)
            else:
                t.add_input(prev_txid=ti['txid'], output_n=ti['vout'],
                            unlocking_script=ti['scriptsig'], value=ti['prevout']['value'],
                            address=ti['prevout'].get('scriptpubkey_address', ''),
                            locking_script=ti['prevout']['scriptpubkey'], sequence=ti['sequence'],
                            witnesses=None if 'witness' not in ti else [bytes.fromhex(w) for w in ti['witness']],
                            strict=self.strict)
        for to in tx['vout']:
            t.add_output(value=to['value'], address=to.get('scriptpubkey_address', ''), spent=None,
                         lock_script=to['scriptpubkey'], strict=self.strict)
        t.update_totals()
        return t

    def gettransaction(self, txid):
        tx = self.compose_request('tx', txid)
        return self._parse_transaction(tx)

    def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS):
        prtxs = []
        before_txid = ''
        while True:
            txs = self.compose_request('address', address, 'txs', before_txid)
            prtxs += txs
            if len(txs) == 25:
                before_txid = txs[-1:][0]['txid']
            else:
                break
            if len(prtxs) > 100:
                break
        txs = []
        for tx in prtxs[::-1]:
            t = self._parse_transaction(tx)
            if t:
                txs.append(t)
            if t.txid == after_txid:
                txs = []
            if len(txs) > limit:
                break
        return txs[:limit]

    def getrawtransaction(self, txid):
        return self.compose_request('tx', txid, 'hex')

    def sendrawtransaction(self, rawtx):
        res = self.compose_request('tx', post_data=rawtx, method='post')
        _logger.debug('mempool response: %s', res)
        return {
            'txid': res,
            'response_dict': {}
        }

    def estimatefee(self, blocks):
        estimates = self.compose_request('v1/fees', 'recommended')
        if blocks < 2:
            return estimates['fastestFee'] * 1000
        elif blocks < 4:
            return estimates['halfHourFee'] * 1000
        if blocks < 7:
            return estimates['hourFee'] * 1000
        else:
            return estimates['minimumFee'] * 1000

    def blockcount(self):
        res = self.compose_request('blocks', 'tip', 'height')
        return res

    def mempool(self, txid=''):
        txids = self.compose_request('mempool', 'txids')
        if not txid:
            return txids
        if txid in txids:
            return [txid]
        return []

    def getblock(self, blockid, parse_transactions, page, limit):
        if isinstance(blockid, int):
            blockid = self.compose_request('block-height', str(blockid))
        if (page == 1 and limit == 10) or limit > 25:
            limit = 25
        bd = self.compose_request('block', blockid)
        btxs = self.compose_request('block', blockid, 'txs', str((page-1)*limit))
        if parse_transactions:
            txs = []
            for tx in btxs[:limit]:
                txs.append(self._parse_transaction(tx))
        else:
            txs = [tx['txid'] for tx in btxs]

        block = {
            'bits': bd['bits'],
            'depth': None,
            'block_hash': bd['id'],
            'height': bd['height'],
            'merkle_root': bd['merkle_root'],
            'nonce': bd['nonce'],
            'prev_block': bd['previousblockhash'],
            'time': bd['timestamp'],
            'tx_count': bd['tx_count'],
            'txs': txs,
            'version': bd['version'],
            'page': page,
            'pages': None if not limit else int(bd['tx_count'] // limit) + (bd['tx_count'] % limit > 0),
            'limit': limit
        }
        return block

    def getrawblock(self, blockid):
        if isinstance(blockid, int):
            blockid = self.compose_request('block-height', str(blockid))
        return self.compose_request('block', blockid, 'raw').hex()

    def isspent(self, txid, output_n):
        res = self.compose_request('tx', txid, 'outspend', str(output_n))
        return 1 if res['spent'] else 0

    # def getinfo(self):

1	# -- coding: utf-8 --
2	#
3	# BitcoinLib - Python Cryptocurrency Library
4	# mempool.space client
5	# © 2021-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, timezone	2✔
23	from bitcoinlib.main import MAX_TRANSACTIONS	2✔
24	from bitcoinlib.services.baseclient import BaseClient, ClientError	2✔
25	from bitcoinlib.transactions import Transaction	2✔
26	from bitcoinlib.encoding import varstr	2✔
27
28	PROVIDERNAME = 'mempool'	2✔
29	# Please note: In the Blockstream API, the first couple of Bitcoin blocks are not correctly indexed,
30	# so transactions from these blocks are missing.
31
32	_logger = logging.getLogger(__name__)	2✔
33
34
35	class MempoolClient(BaseClient):	2✔
36
37	def __init__(self, network, base_url, denominator, *args):	2✔
38	super(self.__class__, self).__init__(network, PROVIDERNAME, base_url, denominator, *args)	2✔
39
40	def compose_request(self, function, data='', parameter='', parameter2='', variables=None, post_data='',	2✔
41	method='get'):
42	url_path = function	2✔
43	if data:	2✔
44	url_path += '/' + data	2✔
45	if parameter:	2✔
46	url_path += '/' + parameter	2✔
47	if parameter2:	2✔
48	url_path += '/' + parameter2	2✔
49	if variables is None:	2✔
50	variables = {}	2✔
51	if self.api_key:	2✔
52	variables.update({'token': self.api_key})	×
53	return self.request(url_path, variables, method, post_data=post_data)	2✔
54
55	def getbalance(self, addresslist):	2✔
56	balance = 0	1✔
57	for address in addresslist:	1✔
58	res = self.compose_request('address', address)	1✔
59	balance += res['chain_stats']['funded_txo_sum'] - res['chain_stats']['spent_txo_sum']	1✔
60	return balance	1✔
61
62	def getutxos(self, address, after_txid='', limit=MAX_TRANSACTIONS):	2✔
63	self.latest_block = self.blockcount() if not self.latest_block else self.latest_block	2✔
64	res = self.compose_request('address', address, 'utxo')	2✔
65	utxos = []	2✔
66	# # key=lambda k: (k[2], pow(10, 20)-k[0].transaction_id, k[3]), reverse=True
67	res = sorted(res, key=lambda k: 0 if 'block_height' not in k['status'] else k['status']['block_height'])	2✔
68	for a in res:	2✔
69	confirmations = 0	2✔
70	block_height = None	2✔
71	if 'block_height' in a['status']:	2✔
72	block_height = a['status']['block_height']	2✔
73	confirmations = self.latest_block - block_height	2✔
74	utxos.append({	2✔
75	'address': address,
76	'txid': a['txid'],
77	'confirmations': confirmations,
78	'output_n': a['vout'],
79	'input_n': 0,
80	'block_height': block_height,
81	'fee': None,
82	'size': 0,
83	'value': a['value'],
84	'script': '',
85	'date': None if 'block_time' not in a['status'] else
86	datetime.fromtimestamp(a['status']['block_time'], timezone.utc)
87	})
88	if a['txid'] == after_txid:	2✔
89	utxos = []	2✔
90	return utxos[:limit]	2✔
91
92	def _parse_transaction(self, tx):	2✔
93	block_height = None if 'block_height' not in tx['status'] else tx['status']['block_height']	2✔
94	confirmations = 0	2✔
95	tx_date = None	2✔
96	status = 'unconfirmed'	2✔
97	if tx['status']['confirmed']:	2✔
98	if block_height:	2✔
99	self.latest_block = self.blockcount() if not self.latest_block else self.latest_block	2✔
100	confirmations = self.latest_block - block_height + 1	2✔
101	tx_date = datetime.fromtimestamp(tx['status']['block_time'], timezone.utc)	2✔
102	status = 'confirmed'	2✔
103	witness_type = 'legacy'	2✔
104	if tx['size'] * 4 > tx['weight']:	2✔
105	witness_type = 'segwit'	2✔
106
107	t = Transaction(locktime=tx['locktime'], version=tx['version'], network=self.network, block_height=block_height,	2✔
108	fee=tx['fee'], size=tx['size'], txid=tx['txid'], date=tx_date, confirmations=confirmations,
109	status=status, coinbase=tx['vin'][0]['is_coinbase'], witness_type=witness_type)
110	for ti in tx['vin']:	2✔
111	if ti['is_coinbase']:	2✔
112	t.add_input(prev_txid=ti['txid'], output_n=ti['vout'], unlocking_script=ti['scriptsig'], value=0,	2✔
113	sequence=ti['sequence'], strict=self.strict, witness_type=witness_type)
114	else:
115	t.add_input(prev_txid=ti['txid'], output_n=ti['vout'],	2✔
116	unlocking_script=ti['scriptsig'], value=ti['prevout']['value'],
117	address=ti['prevout'].get('scriptpubkey_address', ''),
118	locking_script=ti['prevout']['scriptpubkey'], sequence=ti['sequence'],
119	witnesses=None if 'witness' not in ti else [bytes.fromhex(w) for w in ti['witness']],
120	strict=self.strict)
121	for to in tx['vout']:	2✔
122	t.add_output(value=to['value'], address=to.get('scriptpubkey_address', ''), spent=None,	2✔
123	lock_script=to['scriptpubkey'], strict=self.strict)
124	t.update_totals()	2✔
125	return t	2✔
126
127	def gettransaction(self, txid):	2✔
128	tx = self.compose_request('tx', txid)	2✔
129	return self._parse_transaction(tx)	2✔
130
131	def gettransactions(self, address, after_txid='', limit=MAX_TRANSACTIONS):	2✔
132	prtxs = []	2✔
133	before_txid = ''	2✔
134	while True:	1✔
135	txs = self.compose_request('address', address, 'txs', before_txid)	2✔
136	prtxs += txs	2✔
137	if len(txs) == 25:	2✔
138	before_txid = txs[-1:][0]['txid']	×
139	else:
140	break	1✔
141	if len(prtxs) > 100:	×
142	break	×
143	txs = []	2✔
144	for tx in prtxs[::-1]:	2✔
145	t = self._parse_transaction(tx)	2✔
146	if t:	2✔
147	txs.append(t)	2✔
148	if t.txid == after_txid:	2✔
149	txs = []	2✔
150	if len(txs) > limit:	2✔
151	break	1✔
152	return txs[:limit]	2✔
153
154	def getrawtransaction(self, txid):	2✔
155	return self.compose_request('tx', txid, 'hex')	2✔
156
157	def sendrawtransaction(self, rawtx):	2✔
158	res = self.compose_request('tx', post_data=rawtx, method='post')	1✔
159	_logger.debug('mempool response: %s', res)	×
160	return {	×
161	'txid': res,
162	'response_dict': {}
163	}
164
165	def estimatefee(self, blocks):	2✔
166	estimates = self.compose_request('v1/fees', 'recommended')	2✔
167	if blocks < 2:	2✔
168	return estimates['fastestFee'] * 1000	×
169	elif blocks < 4:	2✔
170	return estimates['halfHourFee'] * 1000	×
171	if blocks < 7:	2✔
172	return estimates['hourFee'] * 1000	2✔
173	else:
174	return estimates['minimumFee'] * 1000	1✔
175
176	def blockcount(self):	2✔
177	res = self.compose_request('blocks', 'tip', 'height')	2✔
178	return res	2✔
179
180	def mempool(self, txid=''):	2✔
181	txids = self.compose_request('mempool', 'txids')	2✔
182	if not txid:	2✔
183	return txids	1✔
184	if txid in txids:	2✔
185	return [txid]	×
186	return []	2✔
187
188	def getblock(self, blockid, parse_transactions, page, limit):	2✔
189	if isinstance(blockid, int):	2✔
190	blockid = self.compose_request('block-height', str(blockid))	2✔
191	if (page == 1 and limit == 10) or limit > 25:	2✔
192	limit = 25	2✔
193	bd = self.compose_request('block', blockid)	2✔
194	btxs = self.compose_request('block', blockid, 'txs', str((page-1)*limit))	2✔
195	if parse_transactions:	2✔
196	txs = []	2✔
197	for tx in btxs[:limit]:	2✔
198	txs.append(self._parse_transaction(tx))	2✔
199	else:
200	txs = [tx['txid'] for tx in btxs]	2✔
201
202	block = {	2✔
203	'bits': bd['bits'],
204	'depth': None,
205	'block_hash': bd['id'],
206	'height': bd['height'],
207	'merkle_root': bd['merkle_root'],
208	'nonce': bd['nonce'],
209	'prev_block': bd['previousblockhash'],
210	'time': bd['timestamp'],
211	'tx_count': bd['tx_count'],
212	'txs': txs,
213	'version': bd['version'],
214	'page': page,
215	'pages': None if not limit else int(bd['tx_count'] // limit) + (bd['tx_count'] % limit > 0),
216	'limit': limit
217	}
218	return block	2✔
219
220	def getrawblock(self, blockid):	2✔
221	if isinstance(blockid, int):	1✔
222	blockid = self.compose_request('block-height', str(blockid))	1✔
223	return self.compose_request('block', blockid, 'raw').hex()	1✔
224
225	def isspent(self, txid, output_n):	2✔
226	res = self.compose_request('tx', txid, 'outspend', str(output_n))	1✔
227	return 1 if res['spent'] else 0	1✔
228
229	# def getinfo(self):

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