Merge pull request #756 from cisagov/CRASM_1073_fstrings_to_format

Replace all f-strings to use .format

Author: schmelz21

backend/scripts/populateCountiesCities/cities.py

@@ -117,7 +117,7 @@ def pull_cities():
                         )
                     time.sleep(1)
                 except Exception as e:
-                    print(f"Error: {e}")
+                    print("Error: {}".format(e))
                     pass
 
         df = pd.DataFrame(holding_pen, columns=["State", "County", "City", "URL"])

backend/scripts/populateCountiesCities/counties.py

@@ -62,7 +62,7 @@ def pull_counties():
                 }
             )
         except Exception as e:
-            print(f"Error: {e}")
+            print("Error: {}".format(e))
             pass
 
         time.sleep(1)

backend/src/xfd_django/xfd_api/api_methods/auth.py

@@ -8,20 +8,20 @@
 
 async def handle_okta_callback(request):
     """POST API LOGIC."""
-    print(f"Request from /auth/okta-callback: {str(request)}")
+    print("Request from /auth/okta-callback: {}".format(str(request)))
     body = await request.json()
-    print(f"Request json from callback: {str(request)}")
-    print(f"Request json from callback: {body}")
-    print(f"Body type: {type(body)}")
+    print("Request json from callback: {}".format(str(request)))
+    print("Request json from callback: {}".format(body))
+    print("Body type: {}".format(type(body)))
     code = body.get("code")
-    print(f"Code: {code}")
+    print("Code: {}".format(code))
     if not code:
         return HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail="Code not found in request body",
         )
     jwt_data = await get_jwt_from_code(code)
-    print(f"JWT Data: {jwt_data}")
+    print("JWT Data: {}".format(jwt_data))
     if jwt_data is None:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,

backend/src/xfd_django/xfd_api/api_methods/domain.py

@@ -167,7 +167,7 @@ def export_domains(domain_search: DomainSearch, current_user):
                     for product in service.products.all():
                         if product.name:
                             product_entry = (
-                                f"{product.name} {product.version}"
+                                "{} {}".format(product.name, product.version)
                                 if product.version
                                 else product.name
                             )
@@ -221,5 +221,5 @@ def export_domains(domain_search: DomainSearch, current_user):
 
     except Exception as e:
         # Log the exception for debugging (optional)
-        print(f"Error exporting domains: {e}")
+        print("Error exporting domains: {}".format(e))
         raise HTTPException(status_code=500, detail=str(e))

backend/src/xfd_django/xfd_api/api_methods/organization.py

@@ -250,7 +250,7 @@ def get_organization(organization_id, current_user):
         raise http_exc
 
     except Exception as e:
-        print(f"An error occurred: {e}")
+        print("An error occurred: {}".format(e))
         raise HTTPException(status_code=500, detail=str(e))
 
 
@@ -715,7 +715,7 @@ def delete_organization(org_id: str, current_user):
         # Return success response
         return {
             "status": "success",
-            "message": f"Organization {org_id} has been deleted successfully.",
+            "message": "Organization {} has been deleted successfully.".format(org_id),
         }
 
     except HTTPException as http_exc:
@@ -1064,7 +1064,7 @@ def search_organizations_task(search_body, current_user: User):
         # Use match_all if searchTerm is empty
         if search_body.searchTerm.strip():
             query_body["query"]["bool"]["must"].append(
-                {"wildcard": {"name": f"*{search_body.searchTerm}*"}}
+                {"wildcard": {"name": "*{}*".format(search_body.searchTerm)}}
             )
         else:
             query_body["query"]["bool"]["must"].append({"match_all": {}})
@@ -1076,7 +1076,7 @@ def search_organizations_task(search_body, current_user: User):
             )
 
         # Log the query for debugging
-        print(f"Query body: {query_body}")
+        print("Query body: {}".format(query_body))
 
         # Execute the search
         search_results = client.search_organizations(query_body)

backend/src/xfd_django/xfd_api/api_methods/proxy.py

@@ -32,13 +32,13 @@ async def proxy_request(
     if cookie_name:
         cookies = manipulate_cookie(request, cookie_name)
         if cookies:
-            headers["Cookie"] = f"{cookie_name}={cookies[cookie_name]}"
+            headers["Cookie"] = "{}={}".format(cookie_name, cookies[cookie_name])
 
     # Make the request to the target URL
     async with httpx.AsyncClient() as client:
         proxy_response = await client.request(
             method=request.method,
-            url=f"{target_url}/{path}",
+            url="{}/{}".format(target_url, path),
             headers=headers,
             params=request.query_params,
             content=await request.body(),

backend/src/xfd_django/xfd_api/api_methods/saved_search.py

@@ -235,7 +235,7 @@ def delete_saved_search(saved_search_id, user):
         return JsonResponse(
             {
                 "status": "success",
-                "message": f"Saved search id:{saved_search_id} deleted.",
+                "message": "Saved search id:{} deleted.".format(saved_search_id),
             }
         )
     except User.DoesNotExist:

backend/src/xfd_django/xfd_api/api_methods/scan.py

@@ -270,7 +270,10 @@ def delete_scan(scan_id: str, current_user):
 
         scan.delete()
 
-        return {"status": "success", "message": f"Scan {scan_id} deleted successfully."}
+        return {
+            "status": "success",
+            "message": "Scan {} deleted successfully.".format(scan_id),
+        }
 
     except HTTPException as http_exc:
         raise http_exc
@@ -297,7 +300,7 @@ def run_scan(scan_id: str, current_user):
         scan.save()
         return {
             "status": "success",
-            "message": f"Scan {scan_id} set to manualRunPending.",
+            "message": "Scan {} set to manualRunPending.".format(scan_id),
         }
 
     except HTTPException as http_exc:
@@ -320,7 +323,7 @@ async def invoke_scheduler(current_user):
         lambda_client = LambdaClient()
 
         # Form the lambda function name using environment variable
-        lambda_function_name = f"{os.getenv('SLS_LAMBDA_PREFIX')}-scheduler"
+        lambda_function_name = "{}-scheduler".format(os.getenv("SLS_LAMBDA_PREFIX"))
         print(lambda_function_name)
 
         # Run the Lambda command

backend/src/xfd_django/xfd_api/api_methods/scan_tasks.py

@@ -37,9 +37,9 @@ def list_scan_tasks(search_data: Optional[ScanTaskSearch], current_user):
 
         # Determine the correct ordering based on the 'order' field
         ordering_field = (
-            f"-{search_data.sort}"
+            "-{}".format(search_data.sort)
             if search_data.order and search_data.order.upper() == "DESC"
-            else search_data.sort
+            else "{}".format(search_data.sort)
         )
 
         # Construct query based on filters
@@ -71,7 +71,7 @@ def list_scan_tasks(search_data: Optional[ScanTaskSearch], current_user):
         for task in qs:
             # Ensure scan is not None before accessing its properties
             if task.scan is None:
-                print(f"Warning: ScanTask {task.id} has no scan associated.")
+                print("Warning: ScanTask {} has no scan associated.".format(task.id))
                 scan_data = None
             else:
                 scan_data = {
@@ -177,7 +177,7 @@ def kill_scan_task(scan_task_id, current_user):
         utc_now = datetime.now(timezone.utc)
         scan_task.status = "failed"
         scan_task.finishedAt = utc_now
-        scan_task.output = f"Manually stopped at {utc_now.isoformat()}"
+        scan_task.output = "Manually stopped at {}".format(utc_now.isoformat())
         scan_task.save()
 
         return {"statusCode": 200, "message": "ScanTask successfully marked as failed."}

backend/src/xfd_django/xfd_api/api_methods/search.py

@@ -67,7 +67,7 @@ async def fetch_all_results(
         try:
             response = client.search_domains(request)
         except Exception as e:
-            print(f"Elasticsearch error: {e}")
+            print("Elasticsearch error: {}".format(e))
             raise HTTPException(status_code=500, detail="Error querying Elasticsearch.")
 
         hits = response.get("hits", {}).get("hits", [])
@@ -97,9 +97,9 @@ def process_results(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
                 if "name" in product:
                     product_name = product["name"].lower()
                     product_version = product.get("version", "")
-                    products[
-                        product_name
-                    ] = f"{product['name']} {product_version}".strip()
+                    products[product_name] = "{} {}".format(
+                        product["name"], product_version
+                    ).strip()
 
         res["products"] = ", ".join(products.values())
         processed_results.append(res)
@@ -204,7 +204,7 @@ async def search_export(search_body: DomainSearchBody, current_user) -> Dict[str
     try:
         csv_url = s3_client.save_csv(csv_content, "domains")
     except Exception as e:
-        print(f"S3 upload error: {e}")
+        print("S3 upload error: {}".format(e))
         raise HTTPException(status_code=500, detail="Error uploading CSV to S3.")
 
     return {"url": csv_url}