cic-internal-integration/apps/cic-eth/cic_eth/runnable/daemons/server.py

110 lines
3.6 KiB
Python
Raw Normal View History

2021-10-25 16:51:45 +02:00
# standard imports
import json
import logging
2021-11-18 13:38:20 +01:00
from os import path
2021-10-25 16:51:45 +02:00
from urllib.parse import parse_qsl, urlparse
2021-11-19 14:50:13 +01:00
from cic_eth.server.celery_helper import call
2021-11-18 13:38:20 +01:00
from cic_eth.server.UWSGIOpenAPIRequest import UWSGIOpenAPIRequest
from openapi_core import create_spec
from openapi_core.validation.request.datatypes import OpenAPIRequest
from openapi_core.validation.request.validators import RequestValidator
from openapi_spec_validator.schemas import read_yaml_file
2021-11-19 14:50:13 +01:00
2021-11-18 13:38:20 +01:00
spec_dict = read_yaml_file(path.join(path.dirname(
__file__), '../../server/openapi/server.yaml'))
spec = create_spec(spec_dict)
2021-10-25 16:51:45 +02:00
logging.basicConfig(level=logging.WARNING)
2021-11-19 14:50:13 +01:00
log = logging.getLogger()
2021-11-18 13:38:20 +01:00
2021-10-25 16:51:45 +02:00
# uwsgi application
def application(env, start_response):
2021-11-19 14:50:13 +01:00
oAPIRequest = UWSGIOpenAPIRequest(env)
2021-11-18 13:38:20 +01:00
validator = RequestValidator(spec)
result = validator.validate(oAPIRequest)
# raise errors if request invalid
# result.raise_for_errors()
if result.errors:
# get list of errors
json_data = json.dumps(list(map(lambda e: str(e), result.errors)))
content = json_data.encode('utf-8')
headers = []
headers.append(('Content-Length', str(len(content))),)
headers.append(('Access-Control-Allow-Origin', '*',))
headers.append(('Content-Type', 'application/json',))
start_response('400 Invalid Request', headers)
return [content]
parsed_url = urlparse(env.get('REQUEST_URI')) # /api
2021-10-25 16:51:45 +02:00
path = parsed_url.path
params = dict(parse_qsl(parsed_url.query))
2021-11-19 14:50:13 +01:00
# Setup Channel to receive the result
2021-10-25 16:51:45 +02:00
if path == '/list':
2021-11-19 14:50:13 +01:00
address = params.pop('address')
print('address', address, )
# address, limit=10
data = call('list', address, **params)
2021-10-25 16:51:45 +02:00
elif path == '/balance':
2021-11-19 14:50:13 +01:00
address = params.pop('address')
token_symbol = params.pop('token_symbol')
data = call('balance', address, token_symbol, **params)
for b in data:
b.update({
"balance_available": int(b['balance_network']) + int(b['balance_incoming']) - int(b['balance_outgoing'])
})
2021-10-25 16:51:45 +02:00
elif path == '/create_account':
2021-11-18 16:03:01 +01:00
data = call('create_account', **params)
2021-11-19 14:50:13 +01:00
elif path == '/refill_gas':
address = params.pop('address')
data = call('refill_gas', address)
2021-10-25 16:51:45 +02:00
elif path == '/ping':
2021-11-18 16:03:01 +01:00
data = call('ping', **params)
2021-11-19 14:50:13 +01:00
2021-10-25 16:51:45 +02:00
elif path == '/transfer':
2021-11-19 14:50:13 +01:00
from_address = params.pop('from_address')
to_address = params.pop('to_address')
value = params.pop('value')
token_symbol = params.pop('token_symbol')
data = call('transfer', from_address, to_address, value, token_symbol)
2021-10-25 16:51:45 +02:00
elif path == '/transfer_from':
2021-11-19 14:50:13 +01:00
from_address = params.pop('from_address')
to_address = params.pop('to_address')
value = params.pop('value')
token_symbol = params.pop('token_symbol')
spender_address = params.pop('spender_address')
data = call('transfer_from', from_address, to_address,
value, token_symbol, spender_address)
2021-10-25 16:51:45 +02:00
elif path == '/token':
2021-11-19 14:50:13 +01:00
token_symbol = params.pop('token_symbol')
data = call('token', token_symbol, **params)
2021-10-25 16:51:45 +02:00
elif path == '/tokens':
2021-11-19 14:50:13 +01:00
token_symbols = params.pop('token_symbols')
data = call('tokens', token_symbols, **params)
2021-10-25 16:51:45 +02:00
elif path == '/default_token':
2021-11-18 16:03:01 +01:00
data = call('default_token', )
2021-10-25 16:51:45 +02:00
json_data = json.dumps(data)
content = json_data.encode('utf-8')
headers = []
headers.append(('Content-Length', str(len(content))),)
headers.append(('Access-Control-Allow-Origin', '*',))
headers.append(('Content-Type', 'application/json',))
start_response('200 OK', headers)
return [content]