Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def getutxos(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
res = self.compose_request('address', address, 'utxo')
blockcount = self.blockcount()
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 = blockcount - block_height
utxos.append({
'address': address,
'tx_hash': a['txid'],
'confirmations': confirmations,
'output_n': a['vout'],
def gettransactions(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
txs = []
res1 = self.compose_request('get_tx_received', address, after_txid)
if res1['status'] != 'success':
raise ClientError("Chainso get_tx_received request unsuccessful, status: %s" % res1['status'])
res2 = self.compose_request('get_tx_spent', address, after_txid)
if res2['status'] != 'success':
raise ClientError("Chainso get_tx_spent request unsuccessful, status: %s" % res2['status'])
res = res1['data']['txs'] + res2['data']['txs']
tx_conf = [(t['txid'], t['confirmations']) for t in res]
tx_conf_sorted = sorted(tx_conf, key=lambda x: x[1], reverse=True)
for tx in tx_conf_sorted[:max_txs]:
t = self.gettransaction(tx[0])
txs.append(t)
return txs
def gettransactions(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
"""
Get all transactions for specified address.
Sorted from old to new, so transactions with highest number of confirmations first.
:param address: Address string
:type address: str
:param after_txid: Transaction ID of last known transaction. Only check for transactions after given tx id. Default: Leave empty to return all transaction. If used only provide a single address
:type after_txid: str
:param max_txs: Maximum number of transactions to return
:type max_txs: int
:return list: List of Transaction objects
"""
if not address:
return []
def getutxos(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
address = self._address_convert(address)
if address.witness_type != 'legacy':
raise ClientError("Provider does not support segwit addresses")
res = self.compose_request('addrs', address.address, variables={'unspentOnly': 1, 'limit': 2000})
transactions = []
if not isinstance(res, list):
res = [res]
for a in res:
if 'txrefs' not in a:
continue
if len(a['txrefs']) > 500:
_logger.warning("BlockCypher: Large number of transactions for address %s, "
"Transaction list may be incomplete" % address)
for tx in a['txrefs']:
if tx['tx_hash'] == after_txid:
break
def getutxos(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
address = self._address_convert(address)
res = self.compose_request('addrs', address.address, 'utxo')
txs = []
for tx in res:
if tx['txid'] == after_txid:
break
txs.append({
'address': address.address_orig,
'tx_hash': tx['txid'],
'confirmations': tx['confirmations'],
'output_n': tx['vout'],
'input_n': 0,
'block_height': tx['height'],
'fee': None,
'size': 0,
'value': tx['satoshis'],
def gettransactions(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
address = self._address_convert(address)
res = self.compose_request('addrs', address.address, 'txs')
txs = []
txs_dict = res['items'][::-1]
if after_txid:
txs_dict = txs_dict[[t['txid'] for t in txs_dict].index(after_txid) + 1:]
for tx in txs_dict[:max_txs]:
if tx['txid'] == after_txid:
break
txs.append(self._convert_to_transaction(tx))
return txs
def getutxos(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
utxos = []
page = 1
while True:
variables = {'mode': 'verbose', 'limit': 50, 'page': page, 'order': '1'}
try:
res = self.compose_request('address', 'transactions', address, variables)
res2 = self.compose_request('address', 'unconfirmed/transactions', address, variables)
except ClientError as e:
if "address not found" in self.resp.text:
return []
else:
raise ClientError(e.msg)
txs = res['data']['list']
txs += res2['data']['list']
for tx in txs:
for outp in tx['vOut']:
def getutxos(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
"""
Get list of unspent outputs (UTXO's) for specified address.
Sorted from old to new, so highest number of confirmations first.
:param address: Address string
:type address: str
:param after_txid: Transaction ID of last known transaction. Only check for utxos after given tx id. Default: Leave empty to return all utxos.
:type after_txid: str
:param max_txs: Maximum number of utxo's to return
:type max_txs: int
:return dict: UTXO's per address
"""
if not isinstance(address, TYPE_TEXT):
raise ServiceError("Address parameter must be of type text")
def gettransactions(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
block_height = self.blockcount()
prtxs = []
before_txid = ''
while True:
parameter = 'txs'
if before_txid:
parameter = 'txs/chain/%s' % before_txid
res = self.compose_request('address', address, parameter)
prtxs += res
if len(res) == 25:
before_txid = res[-1:]['txid']
else:
break
txs = []
for tx in prtxs[::-1]:
t = self._parse_transaction(tx, block_height)
def gettransactions(self, address, after_txid='', max_txs=MAX_TRANSACTIONS):
txids = []
offset = 0
while True:
res = self.compose_request('dashboards/address/', data=address, offset=offset)
addr = res['data'][address]
if not addr['transactions']:
break
txids = addr['transactions'][::-1] + txids
offset += 50
if after_txid:
txids = txids[txids.index(after_txid)+1:]
txs = []
for txid in txids[:max_txs]:
txs.append(self.gettransaction(txid))
return txs