Compare commits

..

2 Commits

Author SHA1 Message Date
mike c5b37519ae Merge pull request 'moved files' (#10) from mike/rpc_update into main
Reviewed-on: http://172.16.100.2:3000/mike/fastapi/pulls/10
2023-04-06 13:42:55 +02:00
Bob 7679555f1e moved files 2023-04-06 14:42:08 +03:00
3 changed files with 38 additions and 4 deletions

View File

@ -59,9 +59,7 @@ def checksumCheck(method, address):
else: else:
return False return False
'''
'''
k = sha3.keccak_256() k = sha3.keccak_256()
print("First 65: " + str(address_type[:-65])) print("First 65: " + str(address_type[:-65]))
print(hash64) print(hash64)
@ -77,7 +75,7 @@ def checksumCheck(method, address):
'''
valid = re.compile(r"^(bc1|[13])[a-zA-HJ-NP-Z0-9]{25,39}$") valid = re.compile(r"^(bc1|[13])[a-zA-HJ-NP-Z0-9]{25,39}$")
if valid.match(address) is None: if valid.match(address) is None:
return False return False
@ -181,3 +179,39 @@ def decode_monero_address(encoded):
#print(binascii.hexlify(keccak256), checksum) #print(binascii.hexlify(keccak256), checksum)
return False return False
# class
class RPCXMR(object):
def __init__(self, url, user, password):
self._session = requests.Session()
self._url = url
self._user = user
self._pass = password
self._headers = {}
def call(self, rpcMethod, params):
payload = json.dumps({"method": rpcMethod, "params": params, "jsonrpc": "2.0"})
tries = 3
hadConnectionFailures = False
while True:
try:
response = self._session.post(self._url, headers=self._headers, data=payload, auth=HTTPDigestAuth(self._user, self._pass), timeout=15)
except requests.exceptions.ConnectionError:
tries -= 1
if tries == 0:
raise Exception('Failed to connect for remote procedure call.')
hadFailedConnections = True
print("Couldn't connect for remote procedure call, will sleep for two seconds and then try again ({} more tries)".format(tries))
#time.sleep(2)
else:
if hadConnectionFailures:
print('Connected for remote procedure call after retry.')
break
if not response.status_code in (200, 500):
raise Exception('RPC connection failure: ' + str(response.status_code) + ' ' + response.reason)
responseJSON = response.json()
if 'error' in responseJSON and responseJSON['error'] != None:
raise Exception('Error in RPC call: ' + str(responseJSON['error']))
return responseJSON['result']