-
Notifications
You must be signed in to change notification settings - Fork 10
/
bitcoin_rpc.py
90 lines (73 loc) · 2.74 KB
/
bitcoin_rpc.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
'''
Implements simple interface to bitcoind's RPC.
'''
import json
import base64
import logger
import geventhttpclient
class BitcoinRPC(object):
def __init__(self, host, port, username, password):
self.client = geventhttpclient.HTTPClient(host, port)
self.credentials = base64.b64encode("%s:%s" % (username, password))
self.headers = {
'Content-Type': 'text/json',
'Authorization': 'Basic %s' % self.credentials,
}
def _call_raw(self, data):
logger.log('rpc', 'req:', data)
res = self.client.post('/', body=data, headers=self.headers).read()
logger.log('rpc', 'res:', res)
return res
def _call(self, method, params):
return self._call_raw(json.dumps({
'jsonrpc': '2.0',
'method': method,
'params': params,
'id': '1',
}))
def _pay_call_raw(self, data):
logger.log('pay', 'req:', data)
res = self.client.post('/', body=data, headers=self.headers).read()
logger.log('pay', 'res:', res)
return res
def _pay_call(self, method, params):
return self._pay_call_raw(json.dumps({
'jsonrpc': '2.0',
'method': method,
'params': params,
'id': '1',
}))
def validateaddress(self, address):
resp = self._pay_call('validateaddress', [address])
return json.loads(resp)['result']['isvalid']
def settxfee(self, amount):
resp = self._call('settxfee', [amount])
return resp
def walletpassphrase(self, passphrase, timeout):
resp = self._call('walletpassphrase', [passphrase, timeout])
return resp
def sendtoaddress(self, address, amount):
resp = self._call('sendtoaddress', [address, amount])
return json.loads(resp)
def getbalance(self, account):
resp = self._pay_call('getbalance', [account])
result = json.loads(resp)
if(result['error']):
logger.log('error','getbalance error %s' % result)
return 0
else:
return result['result']
def sendfrom(self, account, to_address, amount):
resp = self._pay_call('sendfrom', [account, to_address, amount])
result = json.loads(resp)
if(result['error']):
logger.log('error','sendfrom error %s' % result)
return None
else:
return result['result']
def getblock(self, blockhash):
resp = self._call('getblock', [blockhash])
return json.loads(resp)['result']
def gettxout(self, txid, n):
resp = self._call('gettxout', [txid, n])
return json.loads(resp)['result']