Linting

backend/census_historical_migration/throwaway_scripts/reprocess_migration_cli_commands.py

@@ -1,7 +1,7 @@
 import argparse
 import time
 
-from util import (
+from census_historical_migration.throwaway_scripts.util import (
     trigger_migration_workflow,
 )
 import subprocess  # nosec

backend/census_historical_migration/throwaway_scripts/start_process_cli_commands.py

@@ -2,7 +2,9 @@
 import subprocess  # nosec
 import time
 
-from util import trigger_migration_workflow
+from census_historical_migration.throwaway_scripts.util import (
+    trigger_migration_workflow,
+)
 
 # This throwaway script spits out code that can be
 # copy-pasted into a bash script, or directly into the command line.

backend/dissemination/sql/sling/bulk_data_export/create_stream_yaml.py

@@ -7,21 +7,22 @@
 import yaml
 from datetime import datetime
 
-obj = {"streams": dict()}
-
-obj["source"] = "FAC_SNAPSHOT_URI"
-obj["target"] = "BULK_DATA_EXPORT"
-obj["defaults"] = {
-    "target_options": {
-        "format": "csv",
-        "compression": "gzip",
-        "file_max_rows": 0,
-    }
+obj: dict = {
+    "streams": {},
+    "source": "FAC_SNAPSHOT_URI",
+    "target": "BULK_DATA_EXPORT",
+    "defaults": {
+        "target_options": {
+            "format": "csv",
+            "compression": "gzip",
+            "file_max_rows": 0,
+        }
+    },
 }
 
-schema = "public_data_v1_0_0"
+SCHEMA = "public_data_v1_0_0"
 
-tables = [
+TABLES = [
     "additional_eins",
     "additional_ueis",
     "combined",
@@ -39,15 +40,15 @@
 # it will not generate anything where data does not exist.
 # This future-proofs us for a year or two, so we don't have to worry
 # about updating the `sling` script that is created.
-years = range(2016, 2031)
+YEARS = range(2016, 2031)
 
 
-for t in tables:
+for t in TABLES:
     ndx = 0
-    for y in years:
-        obj["streams"][f"{schema}.{t}.{ndx}"] = {
+    for y in YEARS:
+        obj["streams"][f"{SCHEMA}.{t}.{ndx}"] = {
             "object": f"bulk_export/{{MM}}/{y}_{t}.csv",
-            "sql": f"SELECT * FROM {schema}.{t} WHERE audit_year = '{y}'",
+            "sql": f"SELECT * FROM {SCHEMA}.{t} WHERE audit_year = '{y}'",
             "mode": "full-refresh",
             "target_options": {
                 "format": "csv",

backend/dissemination/sql/test_api.py

@@ -95,7 +95,7 @@ def _helper(endpoint, keys):
                 | limit(0, EnvVars.RECORDS_REQUESTED - 1)
                 | api(api_version)
             )
-            r = requests.get(base + f"/{endpoint}", headers=h)
+            r = requests.get(base + f"/{endpoint}", headers=h, timeout=60)
             self.good_resp(r.json(), keys)
 
         return _helper
@@ -117,7 +117,7 @@ def test_api_v1_0_3_not_exist(self):
             print("This schema/API should not exist.")
             self.assertTrue(False)
         except Exception:
-            pass
+            self.assertTrue(True)
 
     def test_api_v1_1_0(self):
         f = self.cons(self.ENV, "api_v1_1_0")

backend/pyproject.toml

@@ -35,11 +35,14 @@ disable = [
 
 [tool.bandit]
 # TDB 2022-12-07 - test_schemas and test_validators use random number generation, which bandit doesn't like
+# 2024-11-14 - create_stream_yaml and create_partition_statements will never be touched by users, yet bandit doesn't like the SQL injection risk.
 exclude_dirs = [
     "audit/cross_validation/test_check_has_federal_awards.py",
     "audit/test_schemas.py",
     "audit/test_validators.py",
     "node_modules",
+    "dissemination/sql/fac-snapshot-db/create_partition_statements.py",
+    "dissemination/sql/sling/bulk_data_export/create_stream_yaml.py",
     "dissemination/test_search.py",
     "dissemination/tests.py",
     ".venv"

backend/util/api_perf_test/perf.py

@@ -138,10 +138,10 @@ def make_ratios(d1, d2):
 
 
 if __name__ == "__main__":
-    results1 = {}
-    results2 = {}
-    results3 = {}
-    results4 = {}
+    results1: dict = {}
+    results2: dict = {}
+    results3: dict = {}
+    results4: dict = {}
 
     results1["api110_by_year"] = fetch_fa_by_year_exp("api_v1_1_0")
     results1["ap110"] = fetch_fa_exp("api_v1_1_0")