120 lines
4.7 KiB
Python
120 lines
4.7 KiB
Python
import logging
|
|
import sys
|
|
from typing import List, Optional, Union
|
|
|
|
from cic_eth.server import cache, converters
|
|
from cic_eth.server.cache import setup_cache
|
|
from cic_eth.server.celery import create_celery_wrapper
|
|
from cic_eth.server.models import (DefaultToken, Token, TokenBalance,
|
|
Transaction)
|
|
from fastapi import FastAPI, Query
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def create_app(celery_wrapper):
|
|
|
|
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_wrapper('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_wrapper('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_wrapper(
|
|
'create_account', password=password, register=register)
|
|
return data
|
|
|
|
# def refill_gas(start_response, query: dict):
|
|
# address = query.pop('address')
|
|
# data = celery_wrapper('refill_gas', address)
|
|
# return data
|
|
|
|
# def ping(start_response, query: dict):
|
|
# data = celery_wrapper('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_wrapper('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_wrapper('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_wrapper('token', token_symbol, proof=proof)
|
|
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_wrapper('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 = celery_wrapper('default_token')
|
|
return data
|
|
|
|
def get_token(token_symbol: str):
|
|
data = celery_wrapper('token', token_symbol)
|
|
return Token.new(data)
|
|
return app
|