62 lines
1.6 KiB
Python
62 lines
1.6 KiB
Python
|
import json
|
||
|
import logging
|
||
|
import sys
|
||
|
import uuid
|
||
|
|
||
|
import redis
|
||
|
from cic_eth.api.api_task import Api
|
||
|
from cic_eth.server.config import config
|
||
|
|
||
|
log = logging.getLogger(__name__)
|
||
|
|
||
|
|
||
|
chain_spec = config.get('CHAIN_SPEC')
|
||
|
celery_queue = config.get('CELERY_QUEUE')
|
||
|
|
||
|
api = Api(
|
||
|
chain_spec,
|
||
|
queue=celery_queue,
|
||
|
)
|
||
|
|
||
|
|
||
|
redis_host = config.get('REDIS_HOST')
|
||
|
redis_port = config.get('REDIS_PORT')
|
||
|
redis_db = config.get('REDIS_DB')
|
||
|
|
||
|
|
||
|
def call(method, *args):
|
||
|
""" Creates a redis channel and calls `cic_eth.api` with the provided `method` and `*args`. Returns the result of the api call
|
||
|
"""
|
||
|
redis_channel = str(uuid.uuid4())
|
||
|
r = redis.Redis(redis_host, redis_port, redis_db)
|
||
|
ps = r.pubsub()
|
||
|
ps.subscribe(redis_channel)
|
||
|
api = Api(
|
||
|
chain_spec,
|
||
|
queue=celery_queue,
|
||
|
callback_param='{}:{}:{}:{}'.format(
|
||
|
redis_host, redis_port, redis_db, redis_channel),
|
||
|
callback_task='cic_eth.callbacks.redis.redis',
|
||
|
callback_queue=celery_queue,
|
||
|
)
|
||
|
log.debug(f"cic_eth.api.{method}({args})")
|
||
|
getattr(api, method)(*args)
|
||
|
ps.get_message()
|
||
|
try:
|
||
|
o = ps.get_message(timeout=config.get('REDIS_TIMEOUT'))
|
||
|
log.debug(f"cic_eth.api.{method}({args})\n {o}")
|
||
|
except TimeoutError as e:
|
||
|
sys.stderr.write(
|
||
|
f'cic_eth.api.{method}({args}) timed out:\n {e}')
|
||
|
sys.exit(1)
|
||
|
|
||
|
ps.unsubscribe()
|
||
|
try:
|
||
|
result = json.loads(o['data'])["result"]
|
||
|
return result
|
||
|
|
||
|
except Exception as e:
|
||
|
sys.stderr.write(
|
||
|
f'Unable to parse Data:\n{o}\n Error:\n{e}')
|
||
|
sys.exit(1)
|