Fix static-scan findings

app/broadcast_areas/create-broadcast-areas-db.py

@@ -1,10 +1,10 @@
 #!/usr/bin/env python
 
 import csv
-import pickle
 import sys
 from math import isclose
 from pathlib import Path
+from pickle import dumps  # nosec B403
 
 import geojson
 from notifications_utils.formatters import formatted_list
@@ -104,8 +104,10 @@ def clean_up_invalid_polygons(polygons, indent="    "):
 
             # Make sure the polygon is now valid, and that we haven’t
             # drastically transformed the polygon by ‘fixing’ it
-            assert fixed_polygon.is_valid
-            assert isclose(fixed_polygon.area, shapely_polygon.area, rel_tol=0.001)
+            if not fixed_polygon.is_valid:
+                raise RuntimeError("Fixed polygon is no longer valid")
+            if not isclose(fixed_polygon.area, shapely_polygon.area, rel_tol=0.001):
+                raise RuntimeError("Fixed polygon moved too much")
 
             print(  # noqa: T201
                 f"{indent}Polygon {index + 1}/{len(polygons)} fixed!"
@@ -158,7 +160,8 @@ def polygons_and_simplified_polygons(feature):
     # Check that the simplification process hasn’t introduced bad data
     for dataset in output:
         for polygon in dataset:
-            assert Polygon(polygon).is_valid
+            if not Polygon(polygon).is_valid:
+                raise RuntimeError('Simplification process introduced bad data')
 
     return output + [simplified.utm_crs]
 
@@ -360,7 +363,7 @@ def _add_electoral_wards(dataset_id):
         except KeyError:
             print("Skipping", ward_code, ward_name)  # noqa: T201
 
-    rtree_index_path.open('wb').write(pickle.dumps(rtree_index))
+    rtree_index_path.open('wb').write(dumps(rtree_index))
     repo.insert_broadcast_areas(areas_to_add, keep_old_polygons)
 
 

app/broadcast_areas/repo.py

@@ -1,11 +1,11 @@
-import json
 import os
-import pickle
+import pickle  # nosec B403 - loads only used with trusted input
 import sqlite3
+from json import dumps, loads
 from pathlib import Path
 
 rtree_index_path = Path(__file__).parent / 'rtree.pickle'
-rtree_index = pickle.loads(rtree_index_path.read_bytes())
+rtree_index = pickle.loads(rtree_index_path.read_bytes())  # nosec B301 - trusted input
 
 
 class BroadcastAreasRepository(object):
@@ -111,7 +111,7 @@ class BroadcastAreasRepository(object):
                 ))
                 if not keep_old_features:
                     conn.execute(features_q, (
-                        id, json.dumps(polygons), json.dumps(simple_polygons), utm_crs
+                        id, dumps(polygons), dumps(simple_polygons), utm_crs
                     ))
 
     def query(self, sql, *args):
@@ -127,13 +127,10 @@ class BroadcastAreasRepository(object):
         return sorted(libraries)
 
     def get_areas(self, area_ids):
-        q = """
-        SELECT id, name, count_of_phones, broadcast_area_library_id
-        FROM broadcast_areas
-        WHERE id IN ({})
-        """.format(",".join("?" * len(area_ids)))
+        q = "SELECT id, name, count_of_phones, broadcast_area_library_id FROM broadcast_areas"
+        where = "WHERE id IN ({})".format(",".join("?" * len(area_ids)))
 
-        results = self.query(q, *area_ids)
+        results = self.query(F"{q} {where}", *area_ids)
 
         areas = [
             (row[0], row[1], row[2], row[3])
@@ -147,13 +144,13 @@ class BroadcastAreasRepository(object):
         SELECT broadcast_areas.id, name, count_of_phones, broadcast_area_library_id, simple_polygons, utm_crs
         FROM broadcast_areas
         JOIN broadcast_area_polygons on broadcast_area_polygons.id = broadcast_areas.id
-        WHERE broadcast_areas.id IN ({})
-        """.format(",".join("?" * len(area_ids)))
+        """
+        where = "WHERE broadcast_areas.id IN ({})".format(",".join("?" * len(area_ids)))
 
-        results = self.query(q, *area_ids)
+        results = self.query(F"{q} {where}", *area_ids)
 
         areas = [
-            (row[0], row[1], row[2], row[3], json.loads(row[4]), row[5])
+            (row[0], row[1], row[2], row[3], loads(row[4]), row[5])
             for row in results
         ]
 
@@ -239,7 +236,7 @@ class BroadcastAreasRepository(object):
 
         results = self.query(q, area_id)
 
-        return json.loads(results[0][0]), results[0][1]
+        return loads(results[0][0]), results[0][1]
 
     def get_simple_polygons_for_area(self, area_id):
         q = """
@@ -250,4 +247,4 @@ class BroadcastAreasRepository(object):
 
         results = self.query(q, area_id)
 
-        return json.loads(results[0][0]), results[0][1]
+        return loads(results[0][0]), results[0][1]