#281 #280 and fix exception for Firebase transfer function: use TransactionAbortedError

Author: Hagellach37

mapswipe_workers/mapswipe_workers/firebase_to_postgres/transfer_results.py

@@ -1,6 +1,5 @@
 import csv
 import io
-import datetime as dt
 import dateutil.parser
 
 from mapswipe_workers import auth
@@ -9,18 +8,18 @@
 
 
 def transfer_results(project_id_list=None):
-    '''
+    """
     Download results from firebase,
     saves them to postgres and then deletes the results in firebase.
     This is implemented as a transactional operation as described in
     the Firebase docs to avoid missing new generated results in
     Firebase during execution of this function.
-    '''
+    """
 
     # Firebase transaction function
     def transfer(current_results):
         if current_results is None:
-            logger.info(f'{project_id}: No results in Firebase')
+            logger.info(f"{project_id}: No results in Firebase")
             return dict()
         else:
             results_user_id_list = get_user_ids_from_results(current_results)
@@ -33,41 +32,51 @@ def transfer(current_results):
 
     if not project_id_list:
         # get project_ids from existing results if no project ids specified
-        project_id_list = fb_db.reference('v2/results/').get(shallow=True)
+        project_id_list = fb_db.reference("v2/results/").get(shallow=True)
         if not project_id_list:
             project_id_list = []
-            logger.info(f'There are no results to transfer.')
+            logger.info(f"There are no results to transfer.")
 
-    # get all project ids from postgres, we will only transfer results for projects we have there
+    # get all project ids from postgres
+    # we will only transfer results for projects we have there
     postgres_project_ids = get_projects_from_postgres()
 
     for project_id in project_id_list:
         if project_id not in postgres_project_ids:
-            logger.info(f'{project_id}: This project is not in postgres. We will not transfer results')
+            logger.info(
+                f"{project_id}: This project is not in postgres."
+                f"We will not transfer results"
+            )
             continue
-        elif 'tutorial' in project_id:
-            logger.info(f'{project_id}: these are results for a tutorial. we will not transfer these')
+        elif "tutorial" in project_id:
+            logger.info(
+                f"{project_id}: these are results for a tutorial."
+                f"we will not transfer these"
+            )
             continue
 
-        logger.info(f'{project_id}: Start transfering results')
+        logger.info(f"{project_id}: Start transfering results")
 
-        results_ref = fb_db.reference(f'v2/results/{project_id}')
+        results_ref = fb_db.reference(f"v2/results/{project_id}")
         truncate_temp_results()
 
         try:
             results_ref.transaction(transfer)
-            logger.info(f'{project_id}: Transfered results to postgres')
-        except fb_db.TransactionError:
+            logger.info(f"{project_id}: Transfered results to postgres")
+        # TransactionAbortedError:A transaction was aborted
+        # after exceeding the maximum number of retries.
+        except fb_db.TransactionAbortedError:
             logger.exception(
-                f'{project_id}: Firebase transaction for transfering results failed to commit'
+                f"{project_id}: Firebase transaction for"
+                f"transferingresults failed to commit"
             )
 
     del fb_db
     return project_id_list
 
 
 def results_to_file(results, projectId):
-    '''
+    """
     Writes results to an in-memory file like object
     formatted as a csv using the buffer module (StringIO).
     This can be then used by the COPY statement of Postgres
@@ -81,32 +90,37 @@ def results_to_file(results, projectId):
     -------
     results_file: io.StingIO
         The results in an StringIO buffer.
-    '''
+    """
     # If csv file is a file object, it should be opened with newline=''
 
-    results_file = io.StringIO('')
+    results_file = io.StringIO("")
 
-    w = csv.writer(
-            results_file,
-            delimiter='\t',
-            quotechar="'"
-            )
+    w = csv.writer(results_file, delimiter="\t", quotechar="'")
 
-    logger.info(f'Got %s groups for project {projectId} to transfer' % len(results.items()))
+    logger.info(
+        f"Got %s groups for project {projectId} to transfer" % len(results.items())
+    )
     for groupId, users in results.items():
         for userId, results in users.items():
 
-            # check if all attributes are set, if not don't transfer the results for this group
+            # check if all attributes are set,
+            # if not don't transfer the results for this group
             try:
-                timestamp = results['timestamp']
-                start_time = results['startTime']
-                end_time = results['endTime']
-                results = results['results']
+                timestamp = results["timestamp"]
+                start_time = results["startTime"]
+                end_time = results["endTime"]
+                results = results["results"]
             except KeyError as e:
                 sentry.capture_exception_sentry(e)
-                sentry.capture_message_sentry(f'at least one missing attribute for: {projectId}/{groupId}/{userId}, will skip this one')
+                sentry.capture_message_sentry(
+                    f"at least one missing attribute for:"
+                    f"{projectId}/{groupId}/{userId}, will skip this one"
+                )
                 logger.exception(e)
-                logger.warning(f'at least one missing attribute for: {projectId}/{groupId}/{userId}, will skip this one')
+                logger.warning(
+                    f"at least one missing attribute for:"
+                    f"{projectId}/{groupId}/{userId}, will skip this one"
+                )
                 continue
 
             timestamp = dateutil.parser.parse(timestamp)
@@ -115,16 +129,18 @@ def results_to_file(results, projectId):
 
             if type(results) is dict:
                 for taskId, result in results.items():
-                    w.writerow([
-                        projectId,
-                        groupId,
-                        userId,
-                        taskId,
-                        timestamp,
-                        start_time,
-                        end_time,
-                        result,
-                        ])
+                    w.writerow(
+                        [
+                            projectId,
+                            groupId,
+                            userId,
+                            taskId,
+                            timestamp,
+                            start_time,
+                            end_time,
+                            result,
+                        ]
+                    )
             elif type(results) is list:
                 # TODO: optimize for performance
                 # (make sure data from firebase is always a dict)
@@ -135,16 +151,18 @@ def results_to_file(results, projectId):
                     if result is None:
                         continue
                     else:
-                        w.writerow([
-                            projectId,
-                            groupId,
-                            userId,
-                            taskId,
-                            timestamp,
-                            start_time,
-                            end_time,
-                            result,
-                            ])
+                        w.writerow(
+                            [
+                                projectId,
+                                groupId,
+                                userId,
+                                taskId,
+                                timestamp,
+                                start_time,
+                                end_time,
+                                result,
+                            ]
+                        )
             else:
                 raise TypeError
 
@@ -153,54 +171,54 @@ def results_to_file(results, projectId):
 
 
 def save_results_to_postgres(results_file):
-    '''
+    """
     Saves results to a temporary table in postgres using the COPY Statement of Postgres
     for a more efficient import into the database.
 
     Parameters
     ----------
     results_file: io.StringIO
-    '''
+    """
 
     p_con = auth.postgresDB()
     columns = [
-            'project_id',
-            'group_id',
-            'user_id',
-            'task_id',
-            'timestamp',
-            'start_time',
-            'end_time',
-            'result',
-            ]
-    p_con.copy_from(results_file, 'results_temp', columns)
+        "project_id",
+        "group_id",
+        "user_id",
+        "task_id",
+        "timestamp",
+        "start_time",
+        "end_time",
+        "result",
+    ]
+    p_con.copy_from(results_file, "results_temp", columns)
     results_file.close()
 
-    query_insert_results = '''
+    query_insert_results = """
         INSERT INTO results
             SELECT * FROM results_temp
         ON CONFLICT (project_id,group_id,user_id,task_id)
         DO NOTHING;
-    '''
+    """
     p_con.query(query_insert_results)
-    del(p_con)
+    del p_con
 
 
 def truncate_temp_results():
     p_con = auth.postgresDB()
-    query_truncate_temp_results = '''
+    query_truncate_temp_results = """
                     TRUNCATE results_temp
-                '''
+                """
     p_con.query(query_truncate_temp_results)
     del p_con
 
     return
 
 
 def get_user_ids_from_results(results):
-    '''
+    """
     Get all users based on the ids provided in the results
-    '''
+    """
 
     user_ids = set([])
     for groupId, users in results.items():
@@ -211,14 +229,14 @@ def get_user_ids_from_results(results):
 
 
 def get_projects_from_postgres():
-    '''
+    """
     Get the id of all projects in postgres
-    '''
+    """
 
     pg_db = auth.postgresDB()
-    sql_query = '''
+    sql_query = """
         SELECT project_id from projects;
-    '''
+    """
     raw_ids = pg_db.retr_query(sql_query, None)
     project_ids = [i[0] for i in raw_ids]
 

mapswipe_workers/scripts/archive_old_projects.py

@@ -14,16 +14,32 @@ def get_old_projects():
     return projects
 
 
-def move_project_data_to_v2(project_id, project_data):
+def move_project_data_to_v2(project_id):
     """
     Copy project information from old path to v2/projects in Firebase.
     Add status=archived attribute.
+    Use Firebase transaction function for this.
     """
-    project_data["status"] = "archived"
+
+    # Firebase transaction function
+    def transfer(current_data):
+        current_data["status"] = "archived"
+        current_data["projectType"] = 1
+        fb_db.reference("v2/projects/{0}".format(project_id)).set(current_data)
+        return dict()
+
     fb_db = auth.firebaseDB()
-    fb_db.reference("v2/projects/{0}".format(project_id)).set(project_data)
-    fb_db.reference("projects/{0}".format(project_id)).set({})
-    logger.info(f"moved old project to v2: {project_id}")
+    projects_ref = fb_db.reference(f"projects/{project_id}")
+    try:
+        projects_ref.transaction(transfer)
+        logger.info(f"{project_id}: Transfered project to v2 and delete in old path")
+        return True
+    except fb_db.TransactionAbortedError:
+        logger.exception(
+            f"{project_id}: Firebase transaction"
+            f"for transferring project failed to commit"
+        )
+        return False
 
 
 def delete_old_groups(project_id):
@@ -59,9 +75,10 @@ def archive_old_projects():
 
     projects = get_old_projects()
     for project_id in projects.keys():
-        project_data = projects[project_id]
-        move_project_data_to_v2(project_id, project_data)
-        delete_old_groups(project_id)
+        if move_project_data_to_v2(project_id):
+            delete_old_groups(project_id)
+        else:
+            logger.info(f"didn't delete project and groups for project: {project_id}")
 
     delete_other_old_data()