Implement migration script with ussd and notify

apps/cic-ussd/cic_ussd/tasks/base.py

@@ -5,9 +5,24 @@ import celery
 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
@@ -17,4 +32,11 @@ class CriticalSQLAlchemyTask(CriticalTask):
     autoretry_for = (
         sqlalchemy.exc.DatabaseError,
         sqlalchemy.exc.TimeoutError,
+        sqlalchemy.exc.ResourceClosedError,
+        ) 
+
+
+class CriticalMetadataTask(CriticalTask):
+    autoretry_for = (
+        MetadataStoreError,
     )

apps/cic-ussd/cic_ussd/tasks/callback_handler.py

@@ -53,6 +53,18 @@ def process_account_creation_callback(self, result: str, url: str, status_code:
             session.add(user)
             session.commit()
             session.close()
+    
+            queue = self.request.delivery_info.get('routing_key')
+            s = celery.signature(
+                    'cic_ussd.tasks.metadata.add_phone_pointer',
+                    [
+                        result,
+                        phone_number,
+                        'pgp',
+                    ],
+                    queue=queue,
+                    )
+            s.apply_async()
 
             # expire cache
             cache.expire(task_id, timedelta(seconds=180))
@@ -65,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):
@@ -118,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):
@@ -161,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
@@ -174,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 = {}
 
@@ -202,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__()

apps/cic-ussd/cic_ussd/tasks/metadata.py

@@ -2,15 +2,18 @@
 import json
 import logging
 
-# third-party imports
+# external imports
 import celery
+from hexathon import strip_0x
 
 # local imports
 from cic_ussd.metadata import blockchain_address_to_metadata_pointer
 from cic_ussd.metadata.user import UserMetadata
+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
@@ -46,3 +49,10 @@ def edit_user_metadata(blockchain_address: str, data: bytes, engine: str):
     identifier = blockchain_address_to_metadata_pointer(blockchain_address=blockchain_address)
     user_metadata_client = UserMetadata(identifier=identifier)
     user_metadata_client.edit(data=data, engine=engine)
+
+
+@celery_app.task(bind=True, base=CriticalMetadataTask)
+def add_phone_pointer(self, blockchain_address: str, phone: str, engine: str):
+    stripped_address = strip_0x(blockchain_address)
+    phone_metadata_client = PhonePointerMetadata(identifier=phone.encode('utf-8'), engine=engine)
+    phone_metadata_client.create(data=stripped_address)

apps/cic-ussd/cic_ussd/tasks/ussd_session.py

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