Merge pull request #530 from mapswipe/handle_foreign_key_violation

handling invalid matches

Author: Hagellach37

.github/workflows/actions.yml

@@ -20,7 +20,7 @@ jobs:
       working-directory: ./mapswipe_workers
       run: |
         python -m pip install --upgrade pip
-        pip install flake8 black==19.10b0 isort
+        pip install flake8 black==22.3.0 isort
     - name: Code style
       working-directory: ./mapswipe_workers
       run: |

mapswipe_workers/.pre-commit-config.yaml

@@ -1,7 +1,7 @@
 
 repos:
   - repo: https://github.com/psf/black
-    rev: 19.10b0
+    rev: 22.3.0
     hooks:
       - id: black
   - repo: https://gitlab.com/pycqa/flake8

mapswipe_workers/mapswipe_workers/firebase_to_postgres/transfer_results.py

@@ -54,7 +54,7 @@ def transfer_results(project_id_list=None):
     return project_id_list_transfered
 
 
-def transfer_results_for_project(project_id, results):
+def transfer_results_for_project(project_id, results, filter_mode: bool = False):
     """Transfer the results for a specific project.
     Save results into an in-memory file.
     Copy the results to postgres.
@@ -90,18 +90,27 @@ def transfer_results_for_project(project_id, results):
         # results at relatively high speed.
         results_file = results_to_file(results, project_id)
         truncate_temp_results()
-        save_results_to_postgres(results_file)
+        save_results_to_postgres(results_file, project_id, filter_mode=filter_mode)
     except psycopg2.errors.ForeignKeyViolation as e:
+
         sentry.capture_exception(e)
         sentry.capture_message(
             "could not transfer results to postgres due to ForeignKeyViolation: "
-            f"{project_id}"
+            f"{project_id}; filter_mode={filter_mode}"
         )
         logger.exception(e)
         logger.warning(
             "could not transfer results to postgres due to ForeignKeyViolation: "
-            f"{project_id}"
+            f"{project_id}; filter_mode={filter_mode}"
         )
+
+        # There is an exception where additional invalid tasks are in a group.
+        # If that happens we arrive here and add the flag filtermode=true
+        # to this function, which could solve the issue in save_results_to_postgres.
+        # If it does not solve the issue we arrive again but
+        # since filtermode is already true, we will not try to transfer results again.
+        if not filter_mode:
+            transfer_results_for_project(project_id, results, filter_mode=True)
     except Exception as e:
         sentry.capture_exception(e)
         sentry.capture_message(f"could not transfer results to postgres: {project_id}")
@@ -259,14 +268,16 @@ def results_to_file(results, projectId):
     return results_file
 
 
-def save_results_to_postgres(results_file):
+def save_results_to_postgres(results_file, project_id, filter_mode: bool):
     """
     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
+    filter_mode: boolean
+        If true, try to filter out invalid results.
     """
 
     p_con = auth.postgresDB()
@@ -283,6 +294,23 @@ def save_results_to_postgres(results_file):
     p_con.copy_from(results_file, "results_temp", columns)
     results_file.close()
 
+    if filter_mode:
+        logger.warn(f"trying to remove invalid tasks from {project_id}.")
+
+        filter_query = """
+            DELETE FROM results_temp
+            where task_id in (
+                select task_id from results_temp where task_id not in (
+                    select r.task_id from results_temp r join (
+                        select * from tasks where project_id = %(project_id)s
+                    ) as t
+                    on r.group_id = t.group_id
+                    and r.task_id = t.task_id
+                )
+            )
+        """
+        p_con.query(filter_query, {"project_id": project_id})
+
     query_insert_results = """
         INSERT INTO results
             SELECT * FROM results_temp

mapswipe_workers/mapswipe_workers/generate_stats/project_stats.py

@@ -335,7 +335,9 @@ def get_agg_results_by_task_id(
     # add task geometry using left join
     tasks_df.drop(columns=["project_id", "group_id"], inplace=True)
     agg_results_df = results_by_task_id_df.merge(
-        tasks_df, left_on="task_id", right_on="task_id",
+        tasks_df,
+        left_on="task_id",
+        right_on="task_id",
     )
     logger.info("added geometry to aggregated results")
 

mapswipe_workers/mapswipe_workers/mapswipe_workers.py

@@ -331,7 +331,10 @@ def run_create_tutorials() -> None:
 
 @cli.command("archive")
 @click.option(
-    "--project-id", "-i", help=("Archive project with giving project id"), type=str,
+    "--project-id",
+    "-i",
+    help=("Archive project with giving project id"),
+    type=str,
 )
 @click.option(
     "--project-ids",
@@ -359,7 +362,10 @@ def run_archive_project(project_id, project_ids):
 
 @cli.command("delete")
 @click.option(
-    "--project-id", "-i", help=("Delete project with giving project id"), type=str,
+    "--project-id",
+    "-i",
+    help=("Delete project with giving project id"),
+    type=str,
 )
 @click.option(
     "--project-ids",

mapswipe_workers/mapswipe_workers/project_types/arbitrary_geometry/group.py

@@ -8,7 +8,11 @@ class Group(BaseGroup):
     def __init__(self, project: object, groupId: int) -> None:
         super().__init__(project, groupId)
 
-    def create_tasks(self, feature_ids: List, features: List,) -> None:
+    def create_tasks(
+        self,
+        feature_ids: List,
+        features: List,
+    ) -> None:
         """Create tasks for a group
 
         feature_geometries is a list of geometries or feature in geojson format.

mapswipe_workers/mapswipe_workers/project_types/arbitrary_geometry/task.py

@@ -7,7 +7,10 @@
 
 class Task(BaseTask):
     def __init__(
-        self, group: object, featureId: Union[int, str], feature: Dict,
+        self,
+        group: object,
+        featureId: Union[int, str],
+        feature: Dict,
     ):
         """
         Parameters

mapswipe_workers/mapswipe_workers/project_types/base/project.py

@@ -136,7 +136,9 @@ def save_project(self):
         # Make sure projects get saved in Postgres and Firebase successful
         try:
             self.save_to_postgres(
-                project, groups, groupsOfTasks,
+                project,
+                groups,
+                groupsOfTasks,
             )
             logger.info(
                 f"{self.projectId}" f" - the project has been saved" f" to postgres"
@@ -168,7 +170,9 @@ def save_project(self):
 
         try:
             self.save_to_firebase(
-                project, groups, groupsOfTasks,
+                project,
+                groups,
+                groupsOfTasks,
             )
             logger.info(
                 f"{self.projectId}" f" - the project has been saved" f" to firebase"
@@ -493,7 +497,10 @@ def create_groups_txt_file(self, groups):
             "project_type_specifics",
         )
         w = csv.DictWriter(
-            groups_txt_file, fieldnames=fieldnames, delimiter="\t", quotechar="'",
+            groups_txt_file,
+            fieldnames=fieldnames,
+            delimiter="\t",
+            quotechar="'",
         )
 
         for groupId, group in groups.items():
@@ -575,7 +582,10 @@ def create_tasks_txt_file(self, groupsOfTasks):
             "project_type_specifics",
         )
         w = csv.DictWriter(
-            tasks_txt_file, fieldnames=fieldnames, delimiter="\t", quotechar="'",
+            tasks_txt_file,
+            fieldnames=fieldnames,
+            delimiter="\t",
+            quotechar="'",
         )
 
         for groupId, tasks in groupsOfTasks.items():

mapswipe_workers/mapswipe_workers/utils/geojson_functions.py

@@ -206,7 +206,11 @@ def create_geojson_file_from_dict(final_groups_dict, outfile):
 
     dataSource = driver.CreateDataSource(outfile_temp)
     # create layer
-    layer = dataSource.CreateLayer(outfile_temp, srs, geom_type=ogr.wkbPolygon,)
+    layer = dataSource.CreateLayer(
+        outfile_temp,
+        srs,
+        geom_type=ogr.wkbPolygon,
+    )
 
     # create fields
     field_id = ogr.FieldDefn("group_id", ogr.OFTInteger)
@@ -295,7 +299,11 @@ def create_geojson_file(geometries, outfile):
 
     dataSource = driver.CreateDataSource(outfile_temp)
     # create layer
-    layer = dataSource.CreateLayer(outfile_temp, srs, geom_type=ogr.wkbPolygon,)
+    layer = dataSource.CreateLayer(
+        outfile_temp,
+        srs,
+        geom_type=ogr.wkbPolygon,
+    )
 
     # create fields
     field_id = ogr.FieldDefn("id", ogr.OFTInteger)

mapswipe_workers/mapswipe_workers/utils/tile_functions.py

@@ -96,7 +96,10 @@ def tile_coords_zoom_and_tileserver_to_url(
         # https://www.maptiler.com/google-maps-coordinates-tile-bounds-projection/
         tile_y = int(math.pow(2, tile_z) - tile_y) - 1
         url = tile_server["url"].format(
-            key=tile_server["apiKey"], x=tile_x, y=tile_y, z=tile_z,
+            key=tile_server["apiKey"],
+            x=tile_x,
+            y=tile_y,
+            z=tile_z,
         )
     elif "{-y}" in tile_server["url"]:
         # this uses not the standard TMS tile y coordinate,
@@ -105,10 +108,18 @@ def tile_coords_zoom_and_tileserver_to_url(
         # https://www.maptiler.com/google-maps-coordinates-tile-bounds-projection/
         tile_y = int(math.pow(2, tile_z) - tile_y) - 1
         url = tile_server["url"].replace("{-y}", "{y}")
-        url = url.format(key=tile_server["apiKey"], x=tile_x, y=tile_y, z=tile_z,)
+        url = url.format(
+            key=tile_server["apiKey"],
+            x=tile_x,
+            y=tile_y,
+            z=tile_z,
+        )
     else:
         url = tile_server["url"].format(
-            key=tile_server["apiKey"], x=tile_x, y=tile_y, z=tile_z,
+            key=tile_server["apiKey"],
+            x=tile_x,
+            y=tile_y,
+            z=tile_z,
         )
 
     return url