kamikazechaser/cic-internal-integration
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Ensure serial send of txs to node

Browse Source
This commit is contained in:
nolash
2021-04-08 07:09:38 +02:00
parent 2f0a95d2b6
commit 8f2d3997d3
23 changed files with 261 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
apps/cic-ussd/.config/database.ini
View File

@@ -6,3 +6,5 @@ HOST=localhost
PORT=5432
ENGINE=postgresql
DRIVER=psycopg2
DEBUG=0
POOL_SIZE=1

110
apps/cic-ussd/cic_ussd/db/models/base.py
View File

@@ -1,47 +1,129 @@
# standard imports
# stanard imports
import logging
import datetime
# third-party imports
# external imports
from sqlalchemy import Column, Integer, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import (
StaticPool,
QueuePool,
AssertionPool,
NullPool,
)
logg = logging.getLogger().getChild(__name__)
Model = declarative_base(name='Model')
class SessionBase(Model):
"""The base object for all SQLAlchemy enabled models. All other models must extend this.
"""
__abstract__ = True
id = Column(Integer, primary_key=True)
created = Column(DateTime, default=datetime.datetime.utcnow)
updated = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
id = Column(Integer, primary_key=True)
engine = None
session = None
query = None
"""Database connection engine of the running aplication"""
sessionmaker = None
"""Factory object responsible for creating sessions from the connection pool"""
transactional = True
"""Whether the database backend supports query transactions. Should be explicitly set by initialization code"""
poolable = True
"""Whether the database backend supports connection pools. Should be explicitly set by initialization code"""
procedural = True
"""Whether the database backend supports stored procedures"""
localsessions = {}
"""Contains dictionary of sessions initiated by db model components"""
@staticmethod
def create_session():
session = sessionmaker(bind=SessionBase.engine)
return session()
"""Creates a new database session.
"""
return SessionBase.sessionmaker()
@staticmethod
def _set_engine(engine):
"""Sets the database engine static property
"""
SessionBase.engine = engine
SessionBase.sessionmaker = sessionmaker(bind=SessionBase.engine)
@staticmethod
def build():
Model.metadata.create_all(bind=SessionBase.engine)
def connect(dsn, pool_size=16, debug=False):
"""Create new database connection engine and connect to database backend.
:param dsn: DSN string defining connection.
:type dsn: str
"""
e = None
if SessionBase.poolable:
poolclass = QueuePool
if pool_size > 1:
logg.info('db using queue pool')
e = create_engine(
dsn,
max_overflow=pool_size*3,
pool_pre_ping=True,
pool_size=pool_size,
pool_recycle=60,
poolclass=poolclass,
echo=debug,
)
else:
if pool_size == 0:
poolclass = NullPool
elif debug:
poolclass = AssertionPool
else:
poolclass = StaticPool
e = create_engine(
dsn,
poolclass=poolclass,
echo=debug,
)
else:
logg.info('db connection not poolable')
e = create_engine(
dsn,
echo=debug,
)
SessionBase._set_engine(e)
@staticmethod
# https://docs.sqlalchemy.org/en/13/core/pooling.html#pool-disconnects
def connect(data_source_name):
engine = create_engine(data_source_name, pool_pre_ping=True)
SessionBase._set_engine(engine)
@staticmethod
def disconnect():
"""Disconnect from database and free resources.
"""
SessionBase.engine.dispose()
SessionBase.engine = None
@staticmethod
def bind_session(session=None):
localsession = session
if localsession == None:
localsession = SessionBase.create_session()
localsession_key = str(id(localsession))
logg.debug('creating new session {}'.format(localsession_key))
SessionBase.localsessions[localsession_key] = localsession
return localsession
@staticmethod
def release_session(session=None):
session_key = str(id(session))
if SessionBase.localsessions.get(session_key) != None:
logg.debug('commit and destroy session {}'.format(session_key))
session.commit()
session.close()

6
apps/cic-ussd/cic_ussd/metadata/phone.py
View File

@@ -63,8 +63,9 @@ class PhonePointerMetadata(Metadata):
cic_meta_signer = Signer()
signature = cic_meta_signer.sign_digest(data=data)
algorithm = cic_meta_signer.get_operational_key().get('algo')
decoded_data = data.decode('utf-8')
formatted_data = {
'm': data.decode('utf-8'),
'm': decoded_data,
's': {
'engine': engine,
'algo': algorithm,
@@ -76,8 +77,9 @@ class PhonePointerMetadata(Metadata):
try:
result = make_request(method='PUT', url=self.url, data=formatted_data, headers=self.headers)
logg.info(f'Signed content submission status: {result.status_code}.')
logg.debug(f'signed phone pointer metadata submission status: {result.status_code}.')
result.raise_for_status()
logg.info('phone {} metadata pointer {} set to {}'.format(self.identifier.decode('utf-8'), self.metadata_pointer, decoded_data))
except requests.exceptions.HTTPError as error:
raise MetadataStoreError(error)

2
apps/cic-ussd/cic_ussd/metadata/signer.py
View File

@@ -44,7 +44,7 @@ class Signer:
gpg_keys = self.gpg.list_keys()
key_algorithm = gpg_keys[0].get('algo')
key_id = gpg_keys[0].get("keyid")
logg.info(f'using signing key: {key_id}, algorithm: {key_algorithm}')
logg.debug(f'using signing key: {key_id}, algorithm: {key_algorithm}')
return gpg_keys[0]
def sign_digest(self, data: bytes):

2
apps/cic-ussd/cic_ussd/metadata/user.py
View File

@@ -74,7 +74,7 @@ class UserMetadata(Metadata):
try:
result = make_request(method='PUT', url=self.url, data=formatted_data, headers=self.headers)
logg.info(f'Signed content submission status: {result.status_code}.')
logg.debug(f'signed user metadata submission status: {result.status_code}.')
result.raise_for_status()
except requests.exceptions.HTTPError as error:
raise MetadataStoreError(error)

2
apps/cic-ussd/cic_ussd/runnable/server.py
View File

@@ -65,7 +65,6 @@ config.censor('PASSWORD', 'DATABASE')
# define log levels
if args.vv:
logging.getLogger().setLevel(logging.DEBUG)
logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)
elif args.v:
logging.getLogger().setLevel(logging.INFO)
@@ -182,6 +181,7 @@ def application(env, start_response):
logg.error('invalid phone number {}'.format(phone_number))
start_response('400 Invalid phone number format', errors_headers)
return []
logg.debug('session {} started for {}'.format(external_session_id, phone_number))
# handle menu interaction requests
chain_str = chain_spec.__str__()

3
apps/cic-ussd/cic_ussd/runnable/tasker.py
View File

@@ -20,6 +20,7 @@ from cic_ussd.validator import validate_presence
logging.basicConfig(level=logging.WARNING)
logg = logging.getLogger()
logging.getLogger('gnupg').setLevel(logging.WARNING)
config_directory = '/usr/local/etc/cic-ussd/'
@@ -47,7 +48,7 @@ logg.debug(config)
# connect to database
data_source_name = dsn_from_config(config)
SessionBase.connect(data_source_name=data_source_name)
SessionBase.connect(data_source_name, pool_size=int(config.get('DATABASE_POOL_SIZE')), debug=config.true('DATABASE_DEBUG'))
# verify database connection with minimal sanity query
session = SessionBase.create_session()

19
apps/cic-ussd/cic_ussd/tasks/base.py
View File

@@ -6,9 +6,23 @@ import sqlalchemy
# local imports
from cic_ussd.error import MetadataStoreError
from cic_ussd.db.models.base import SessionBase
class CriticalTask(celery.Task):
class BaseTask(celery.Task):
session_func = SessionBase.create_session
def create_session(self):
return BaseTask.session_func()
def log_banner(self):
logg.debug('task {} root uuid {}'.format(self.__class__.__name__, self.request.root_id))
return
class CriticalTask(BaseTask):
retry_jitter = True
retry_backoff = True
retry_backoff_max = 8
@@ -18,7 +32,8 @@ class CriticalSQLAlchemyTask(CriticalTask):
autoretry_for = (
sqlalchemy.exc.DatabaseError,
sqlalchemy.exc.TimeoutError,
)
sqlalchemy.exc.ResourceClosedError,
)
class CriticalMetadataTask(CriticalTask):

7
apps/cic-ussd/cic_ussd/tasks/callback_handler.py
View File

@@ -77,6 +77,8 @@ def process_account_creation_callback(self, result: str, url: str, status_code:
session.close()
raise ActionDataNotFoundError(f'Account creation task: {task_id}, returned unexpected response: {status_code}')
session.close()
@celery_app.task
def process_incoming_transfer_callback(result: dict, param: str, status_code: int):
@@ -130,6 +132,7 @@ def process_incoming_transfer_callback(result: dict, param: str, status_code: in
session.close()
raise ValueError(f'Unexpected status code: {status_code}.')
session.close()
@celery_app.task
def process_balances_callback(result: list, param: str, status_code: int):
@@ -173,7 +176,6 @@ def define_transaction_action_tag(
def process_statement_callback(result, param: str, status_code: int):
if status_code == 0:
# create session
session = SessionBase.create_session()
processed_transactions = []
# process transaction data to cache
@@ -186,6 +188,7 @@ def process_statement_callback(result, param: str, status_code: int):
if '0x0000000000000000000000000000000000000000' in source_token:
pass
else:
session = SessionBase.create_session()
# describe a processed transaction
processed_transaction = {}
@@ -214,6 +217,8 @@ def process_statement_callback(result, param: str, status_code: int):
else:
logg.warning(f'Tx with recipient not found in cic-ussd')
session.close()
# add transaction values
processed_transaction['to_value'] = from_wei(value=transaction.get('to_value')).__str__()
processed_transaction['from_value'] = from_wei(value=transaction.get('from_value')).__str__()

2
apps/cic-ussd/cic_ussd/tasks/metadata.py
View File

@@ -13,7 +13,7 @@ from cic_ussd.metadata.phone import PhonePointerMetadata
from cic_ussd.tasks.base import CriticalMetadataTask
celery_app = celery.current_app
logg = logging.getLogger()
logg = logging.getLogger().getChild(__name__)
@celery_app.task

1
apps/cic-ussd/cic_ussd/tasks/ussd_session.py
View File

@@ -70,3 +70,4 @@ def persist_session_to_db(external_session_id: str):
session.close()
raise SessionNotFoundError('Session does not exist!')
session.close()

2
apps/cic-ussd/docker/start_tasker.sh
View File

@@ -2,4 +2,4 @@
. /root/db.sh
/usr/local/bin/cic-ussd-tasker -vv "$@"
/usr/local/bin/cic-ussd-tasker $@

4
apps/cic-ussd/docker/start_uwsgi.sh
View File

@@ -2,4 +2,6 @@
. /root/db.sh
/usr/local/bin/uwsgi --wsgi-file /usr/local/lib/python3.8/site-packages/cic_ussd/runnable/server.py --http :9000 --pyargv "-vv"
server_port=${SERVER_PORT:-9000}
/usr/local/bin/uwsgi --wsgi-file /usr/local/lib/python3.8/site-packages/cic_ussd/runnable/server.py --http :$server_port --pyargv "$@"