154 lines
5.1 KiB
Python
154 lines
5.1 KiB
Python
import logging
|
|
import sys
|
|
from typing import List, Optional, Union
|
|
|
|
import redis
|
|
# standard imports
|
|
from cic_eth.server import cache, celery, converters
|
|
from cic_eth.server.config import args, config
|
|
from cic_eth.server.models import (DefaultToken, Token, TokenBalance,
|
|
Transaction)
|
|
# standard imports
|
|
from fastapi import FastAPI, Query
|
|
|
|
# define log levels
|
|
if args.vv:
|
|
logging.getLogger().setLevel(logging.DEBUG)
|
|
elif args.v:
|
|
logging.getLogger().setLevel(logging.INFO)
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# define universal redis cache access
|
|
cache.Cache.store = redis.StrictRedis(host=config.get('REDIS_HOST'), port=config.get(
|
|
'REDIS_PORT'), db=config.get('REDIS_DB'), decode_responses=True)
|
|
|
|
|
|
app = FastAPI(debug=True,
|
|
title="Grassroots Economics",
|
|
description="CIC ETH API",
|
|
version="0.0.1",
|
|
terms_of_service="https://www.grassrootseconomics.org/pages/terms-and-conditions.html",
|
|
contact={
|
|
"name": "Grassroots Economics",
|
|
"url": "https://www.grassrootseconomics.org",
|
|
"email": "will@grassecon.org"
|
|
},
|
|
license_info={
|
|
"name": "GPLv3",
|
|
})
|
|
|
|
|
|
@app.get("/transactions", response_model=List[Transaction])
|
|
def transactions(address: str, limit: Optional[str] = 10):
|
|
return celery.call('list', address, limit=limit)
|
|
|
|
|
|
@app.get("/balance", response_model=List[TokenBalance])
|
|
def balance(token_symbol: str, address: str = Query(..., title="Address", min_length=40, max_length=42), include_pending: bool = True):
|
|
log.info(f"address: {address}")
|
|
log.info(f"token_symbol: {token_symbol}")
|
|
data = celery.call('balance', address, token_symbol,
|
|
include_pending=include_pending)
|
|
for b in data:
|
|
token = get_token(token_symbol)
|
|
b['balance_network'] = converters.from_wei(
|
|
token.decimals, int(b['balance_network']))
|
|
b['balance_incoming'] = converters.from_wei(
|
|
token.decimals, int(b['balance_incoming']))
|
|
b['balance_outgoing'] = converters.from_wei(
|
|
token.decimals, int(b['balance_outgoing']))
|
|
|
|
b.update({
|
|
"balance_available": int(b['balance_network']) + int(b['balance_incoming']) - int(b['balance_outgoing'])
|
|
})
|
|
return data
|
|
|
|
|
|
@app.post("/create_account")
|
|
def create_account(password: Optional[str] = None, register: bool = True):
|
|
data = celery.call('create_account', password=password, register=register)
|
|
return data
|
|
|
|
|
|
# def refill_gas(start_response, query: dict):
|
|
# address = query.pop('address')
|
|
# data = celery.call('refill_gas', address)
|
|
# return data
|
|
|
|
|
|
# def ping(start_response, query: dict):
|
|
# data = celery.call('ping', **query)
|
|
# return data
|
|
|
|
@app.post("/transfer")
|
|
def transfer(from_address: str, to_address: str, value: int, token_symbol: str):
|
|
token = get_token(
|
|
token_symbol)
|
|
wei_value = converters.to_wei(token.decimals, int(value))
|
|
data = celery.call('transfer', from_address,
|
|
to_address, wei_value, token_symbol)
|
|
return data
|
|
|
|
|
|
@app.post("/transfer_from")
|
|
def transfer_from(from_address: str, to_address: str, value: int, token_symbol: str, spender_address: str):
|
|
token = get_token(
|
|
token_symbol)
|
|
wei_value = converters.to_wei(token.decimals, int(value))
|
|
data = celery.call('transfer_from', from_address, to_address,
|
|
wei_value, token_symbol, spender_address)
|
|
return data
|
|
|
|
|
|
@app.get("/token", response_model=Token)
|
|
def token(token_symbol: str, proof: Optional[str] = None):
|
|
token = get_token(token_symbol)
|
|
if token == None:
|
|
sys.stderr.write(f"Cached Token {token_symbol} not found")
|
|
data = celery.call('token', token_symbol, proof=proof)
|
|
cache.set_token_data(token_symbol, data)
|
|
token = Token.new(data)
|
|
sys.stderr.write(f"Token {token}")
|
|
|
|
return token
|
|
|
|
|
|
@app.get("/tokens", response_model=List[Token])
|
|
def tokens(token_symbols: Optional[List[str]] = Query(None), proof: Optional[Union[str, List[str], List[List[str]]]] = None):
|
|
data = celery.call('tokens', token_symbols,
|
|
catch=len(token_symbols), proof=proof)
|
|
if data:
|
|
tokens = []
|
|
for token in data:
|
|
print(f"Token: {token}")
|
|
tokens.append(Token.new(token))
|
|
return tokens
|
|
return None
|
|
|
|
|
|
@app.get("/default_token", response_model=DefaultToken)
|
|
def default_token():
|
|
data = cache.get_default_token()
|
|
if data is None:
|
|
data = celery.call('default_token')
|
|
if data is not None:
|
|
cache.set_default_token(data)
|
|
return data
|
|
|
|
|
|
def get_token(token_symbol: str):
|
|
data = cache.get_token_data(token_symbol)
|
|
log.debug(f"cached token data: {data}")
|
|
if data == None:
|
|
data = celery.call('token', token_symbol)
|
|
log.debug(
|
|
f"No token data setting token data for: {token_symbol} to {data}")
|
|
cache.set_token_data(token_symbol, data)
|
|
return Token.new(data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=5000, log_level="info")
|