-
Notifications
You must be signed in to change notification settings - Fork 2
/
livepeer_lpt_transactions.py
84 lines (61 loc) · 2.5 KB
/
livepeer_lpt_transactions.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
import requests, json, datetime
from flask_restful import Api, Resource, reqparse, request
# Get current ETH block height (in decimal)
def getCurrentEthereumBlock():
try:
# Pings Etherscan for the current block
resp = requests.get('https://api.etherscan.io/api?module=proxy&action=eth_blockNumber&apikey=YourApiKeyToken')
# Convert block number to integer
block = int(resp.json()['result'], 16)
return block
except:
return -1
# Get all transactions for a single address in a block range
def getTxnsForBlockRange(address, startBlock, endBlock):
try:
# Pings Ethererscan for transactions by address
url = 'http://api.etherscan.io/api?module=account&action=txlist&address={}&startblock={}&endblock={}&sort=desc&apikey=YourApiKeyToken'
resp = requests.get(url.format(address, startBlock, endBlock))
txns = resp.json()['result']
return txns
except:
return None
# Return all the transactions given multiple addresses
def getTxnsInLastNDays(addresses, days):
# Get current block number
curBlock = getCurrentEthereumBlock()
# Get starting block number
startBlock = curBlock - int(days * 4 * 60 * 24)
allTxns = []
for address in addresses:
result = getTxnsForBlockRange(address, startBlock, curBlock)
if result is not None:
# Add transactions to allTxns
allTxns.extend(result)
return allTxns
class livepeer_lpt_transactions(Resource):
def get(self, service):
if service not in ['get_all_lpt_transactions_from_multiple_wallets']:
return {'data':service + ' is not a function.', 'code':404}
def parse_params(arguments):
# API call params parser.
parser = reqparse.RequestParser()
for argument in arguments:
parser.add_argument(argument)
return parser
# Declare data variable.
data = {}
if service == "get_all_lpt_transactions_from_multiple_wallets":
# Decode body params from request
arguments = ["addresses", "days"]
parser = parse_params(arguments)
args = parser.parse_args()
# Declare query params as variables
addresses = args["addresses"]
days = args["days"]
# Replace "%2C" with ",". This switch of "," for "%2C" can occur when API is queried.
wallet_addresses = addresses.replace('%2C',',')
# Retrieve the transactions for multiple wallets over the stated number of days.
data = getTxnsInLastNDays([wallet_addresses], float(days))
# Return data as JSON
return data