fix: code quality - variable typo, dead import, PEP 8 identity checks, unnecessary f-strings

application/cmd/cre_main.py

@@ -11,7 +11,6 @@
 from collections import deque
 from typing import Any, Callable, Dict, List, Optional, Tuple, TYPE_CHECKING
 import hashlib
-import json as _json
 from rq import Queue, job, exceptions
 from sqlalchemy import not_
 
@@ -462,7 +461,7 @@ def _standard_structure_fingerprint(resource_name: str) -> str:
                     tuple(sorted(links)),
                 )
             )
-        payload = _json.dumps(
+        payload = json.dumps(
             sorted(rows), sort_keys=True, separators=(",", ":"), ensure_ascii=False
         )
         return hashlib.sha256(payload.encode("utf-8")).hexdigest()
@@ -698,7 +697,7 @@ def download_gap_analysis_from_upstream(cache: str) -> None:
                         tojson = res.json()
                         if "result" not in tojson:
                             continue
-                        payload = _json.dumps({"result": tojson.get("result")})
+                        payload = json.dumps({"result": tojson.get("result")})
                         if not gap_analysis.primary_gap_analysis_payload_is_material(
                             payload
                         ):
@@ -715,7 +714,7 @@ def download_gap_analysis_from_upstream(cache: str) -> None:
                     tojson = res.json()
                     if "result" not in tojson:
                         continue
-                    payload = _json.dumps({"result": tojson.get("result")})
+                    payload = json.dumps({"result": tojson.get("result")})
                     if not gap_analysis.primary_gap_analysis_payload_is_material(
                         payload
                     ):
@@ -1242,13 +1241,13 @@ def populate_neo4j_db(cache: str):
     ):
         logger.info("Skipping Neo4j population as per environment variables")
         return
-    logger.info(f"Populating neo4j DB: Connecting to SQL DB")
+    logger.info("Populating neo4j DB: Connecting to SQL DB")
     database = db_connect(path=cache)
     if database.neo_db:
-        logger.info(f"Populating neo4j DB: Populating")
+        logger.info("Populating neo4j DB: Populating")
         database.neo_db.populate_DB(database.session)
-        logger.info(f"Populating neo4j DB: Complete")
+        logger.info("Populating neo4j DB: Complete")
     else:
         logger.warning(
-            f"Populating neo4j DB: database.neo_db is None, skipping population"
+            "Populating neo4j DB: database.neo_db is None, skipping population"
         )

application/database/db.py

@@ -439,7 +439,7 @@ class NeoDocument(StructuredNode):
 
     @classmethod
     def to_cre_def(self, node, parse_links=True):
-        raise Exception(f"Shouldn't be parsing a NeoDocument")
+        raise Exception("Shouldn't be parsing a NeoDocument")
 
     @classmethod
     def get_links(self, links_dict):
@@ -461,7 +461,7 @@ class NeoNode(NeoDocument):
 
     @classmethod
     def to_cre_def(self, node, parse_links=True):
-        raise Exception(f"Shouldn't be parsing a NeoNode")
+        raise Exception("Shouldn't be parsing a NeoNode")
 
 
 class NeoStandard(NeoNode):
@@ -1980,7 +1980,7 @@ def add_internal_link(
             ltype (cre_defs.LinkTypes, optional): the linktype
         Returns: the cre_defs.Link or None in case of error (cycle)
         """
-        if ltype == None:
+        if ltype is None:
             raise ValueError("Every link should have a link type")
 
         if ltype == cre_defs.LinkTypes.PartOf:
@@ -2908,7 +2908,7 @@ def gap_analysis(
         key = node.id
         if not key:
             logger.error(
-                f"key is empty, this is a bug and this gap analysis will not progress"
+                "key is empty, this is a bug and this gap analysis will not progress"
             )
             continue
         if key not in grouped_paths:
@@ -2934,7 +2934,7 @@ def gap_analysis(
         end_key = path["end"].id
         if not end_key:
             logger.error(
-                f"end_key is empty, this is a bug and this gap analysis will not progress"
+                "end_key is empty, this is a bug and this gap analysis will not progress"
             )
             continue
         path["score"] = get_path_score(path)

application/database/inmemory_graph.py

@@ -81,7 +81,7 @@ def get_hierarchy(self, rootIDs: List[str], creID: str):
         if creID in rootIDs:
             return 0
 
-        if self.__parent_child_subgraph == None:
+        if self.__parent_child_subgraph is None:
             if len(self.__graph.edges) == 0:
                 raise ValueError("Graph has no edges")
             include_cres = []

application/prompt_client/prompt_client.py

@@ -396,7 +396,7 @@ def find_missing_embeddings(self, database: db.Node_collection) -> List[str]:
         Returns:
             List[str]: a list of db ids which do not have embeddings
         """
-        logger.info(f"syncing nodes with embeddings")
+        logger.info("syncing nodes with embeddings")
         missing_embeddings = []
         for doc_type in cre_defs.Credoctypes:
             db_ids = []
@@ -1211,7 +1211,7 @@ def get_id_of_most_similar_cre_paginated(
 
         if max_similarity < similarity_threshold:
             logger.info(
-                f"there is no good cre candidate for this standard section, returning nothing"
+                "there is no good cre candidate for this standard section, returning nothing"
             )
             return None, None
         return most_similar_id, max_similarity

application/utils/external_project_parsers/parsers/export_format_parser.py

@@ -85,7 +85,7 @@ def parse_export_format(lfile: List[Dict[str, Any]]) -> Dict[str, List[defs.Docu
                         logger.warning(
                             f"Link between {highest_cre.name} and {working_cre.name} already exists"
                         )
-                elif highest_cre == None:
+                elif highest_cre is None:
                     highest_cre = working_cre
                     highest_index = i
 

application/utils/gap_analysis.py

@@ -97,9 +97,9 @@ def get_path_score(path):
 
         if step["relationship"] == "CONTAINS":
             penalty_type = f"CONTAINS_{get_relation_direction(step, previous_id)}"
-        pentalty = PENALTIES[penalty_type]
-        score += pentalty
-        step["score"] = pentalty
+        penalty = PENALTIES[penalty_type]
+        score += penalty
+        step["score"] = penalty
         previous_id = get_next_id(step, previous_id)
     return score
 

application/utils/oscal_utils.py

@@ -47,7 +47,7 @@ def document_to_oscal(
             version=version,
             links=[common.Link(href=hyperlink)],
         )
-    if uuid == None or uuid == "":
+    if uuid is None or uuid == "":
         uuid = str(uuid4())
     c = catalog.Catalog(metadata=m, uuid=uuid)
     controls: List[catalog.Control] = []

application/utils/spreadsheet_parsers.py

@@ -290,7 +290,7 @@ def parse_export_format(lfile: List[Dict[str, Any]]) -> Dict[str, List[defs.Docu
                         logger.warning(
                             f"Link between {highest_cre.name} and {working_cre.name} already exists"
                         )
-                elif highest_cre == None:
+                elif highest_cre is None:
                     highest_cre = working_cre
                     highest_index = i
 
@@ -499,7 +499,7 @@ def parse_hierarchical_export_format(
         current_hierarchy, name = get_highest_cre_name(
             mapping=mapping, highest_hierarchy=max_hierarchy
         )
-        if name == None:  # skip empty lines
+        if name is None:  # skip empty lines
             continue
 
         if current_hierarchy > 0:  # find the previous higher CRE so we can link

application/web/web_main.py

@@ -210,7 +210,7 @@ def find_node_by_name(
     sectionID: str = "",
 ) -> Any:
     if posthog:
-        posthog.capture(f"find_node_by_name", f"name:{name};nodeType{ntype}")
+        posthog.capture("find_node_by_name", f"name:{name};nodeType{ntype}")
 
     database = db.Node_collection()
     opt_section = section or request.args.get("section")
@@ -302,7 +302,7 @@ def find_node_by_name(
 def find_document_by_tag() -> Any:
     tags = request.args.getlist("tag")
     if posthog:
-        posthog.capture(f"find_document_by_tag", f"tags:{tags}")
+        posthog.capture("find_document_by_tag", f"tags:{tags}")
 
     database = db.Node_collection()
     # opt_osib = request.args.get("osib")
@@ -341,7 +341,7 @@ def find_document_by_tag() -> Any:
 def map_analysis() -> Any:
     standards = request.args.getlist("standard")
     if posthog:
-        posthog.capture(f"map_analysis", f"standards:{standards}")
+        posthog.capture("map_analysis", f"standards:{standards}")
 
     database = db.Node_collection()
     if len(standards) < 2:
@@ -447,7 +447,7 @@ def map_analysis() -> Any:
 def map_analysis_weak_links() -> Any:
     standards = request.args.getlist("standard")
     if posthog:
-        posthog.capture(f"map_analysis_weak_links", f"standards:{standards}")
+        posthog.capture("map_analysis_weak_links", f"standards:{standards}")
 
     key = request.args.get("key")
     cache_key = gap_analysis.make_subresources_key(standards=standards, key=key)
@@ -535,7 +535,7 @@ def fetch_job() -> Any:
 @app.route("/rest/v1/standards", methods=["GET"])
 def standards() -> Any:
     if posthog:
-        posthog.capture(f"standards", "")
+        posthog.capture("standards", "")
 
     database = db.Node_collection()
     standards = list(database.standards())
@@ -603,7 +603,7 @@ def text_search() -> Any:
     if not text:
         return jsonify({"error": "text parameter is required"}), 400
     if posthog:
-        posthog.capture(f"text_search", f"text:{text}")
+        posthog.capture("text_search", f"text:{text}")
 
     opt_format = request.args.get("format")
     documents = database.text_search(text)
@@ -663,7 +663,7 @@ def find_root_cres() -> Any:
 
     """
     if posthog:
-        posthog.capture(f"find_root_cres", "")
+        posthog.capture("find_root_cres", "")
 
     database = db.Node_collection()
     # opt_osib = request.args.get("osib")
@@ -744,7 +744,7 @@ def smartlink(
     # ATTENTION: DO NOT MESS WITH THIS FUNCTIONALITY WITHOUT A TICKET AND CORE CONTRIBUTORS APPROVAL!
     # CRITICAL FUNCTIONALITY DEPENDS ON THIS!
     if posthog:
-        posthog.capture(f"smartlink", f"name:{name}")
+        posthog.capture("smartlink", f"name:{name}")
 
     database = db.Node_collection()
     opt_version = request.args.get("version")
@@ -800,7 +800,7 @@ def smartlink(
         )
         return redirect(redirectors.redirect(name, section))
     else:
-        logger.warning(f"not sure what happened, 404")
+        logger.warning("not sure what happened, 404")
         return abort(404, "Document does not exist")
 
 
@@ -824,7 +824,7 @@ def deeplink(
     opt_version = request.args.get("version")
     opt_subsection = request.args.get("subsection")
     if posthog:
-        posthog.capture(f"deeplink", f"name:{name}")
+        posthog.capture("deeplink", f"name:{name}")
 
     if opt_section:
         opt_section = urllib.parse.unquote(opt_section)
@@ -1077,7 +1077,7 @@ def admin_import_run_apply(run_id: str) -> Any:
 def chat_cre() -> Any:
     message = request.get_json(force=True)
     if posthog:
-        posthog.capture(f"chat_cre", "")
+        posthog.capture("chat_cre", "")
 
     database = db.Node_collection()
     # Lazy import to avoid loading heavy prompt/ML dependencies at web boot.
@@ -1233,7 +1233,7 @@ def logout():
 def all_cres() -> Any:
     database = db.Node_collection()
     if posthog:
-        posthog.capture(f"all_cres", "")
+        posthog.capture("all_cres", "")
 
     page = 1
     per_page = ITEMS_PER_PAGE
@@ -1261,7 +1261,7 @@ def all_cres() -> Any:
 @app.route("/rest/v1/cre_csv", methods=["GET"])
 def get_cre_csv() -> Any:
     if posthog:
-        posthog.capture(f"get_cre_csv", "")
+        posthog.capture("get_cre_csv", "")
 
     database = db.Node_collection()
     root_cres = database.get_root_cres()