Replace print() calls with logging across SDK, examples, tests, and docs

CheckmarxPythonSDK/CxODataApiSDK/Utilities.py

@@ -1,6 +1,8 @@
 import csv
 from itertools import groupby
 from copy import deepcopy
+import logging
+logger = logging.getLogger(__name__)
 
 from .ProjectsODataAPI import (
     get_all_projects_id_name,
@@ -105,8 +107,8 @@ def get_all_results_with_count_for_each_project_json_format(
                     {"ScanId": scan_id, "ResultsWithQuery": results_with_query}
                 )
             except ValueError as e:
-                print(e)
-                print("Fail to fetch data for scan id: {id} ".format(id=scan_id))
+                logger.info(e)
+                logger.error("Fail to fetch data for scan id: {id} ".format(id=scan_id))
 
         project.update({"Scans": results_list})
 
@@ -167,11 +169,9 @@ def get_result(project, filter_false_positive=False, threshold=0):
 
     last_scan_id = get_last_scan_id_of_a_project(project_id=project_id)
     if not last_scan_id:
-        print(
-            "Project name: {name}, id : {id} has no scans".format(
+        logger.info("Project name: {name}, id : {id} has no scans".format(
                 name=project_name, id=project_id
-            )
-        )
+            ))
         return None
 
     last_full_scan_id = get_last_full_scan_id_of_a_project(project_id=project_id)
@@ -181,8 +181,8 @@ def get_result(project, filter_false_positive=False, threshold=0):
             scan_id=last_scan_id
         )
     except ValueError as e:
-        print("Fail to get scan result for scan id: {id}".format(id=last_scan_id))
-        print("Exception: {error} ".format(error=e))
+        logger.error("Fail to get scan result for scan id: {id}".format(id=last_scan_id))
+        logger.error("Exception: {error} ".format(error=e))
         return None
 
     if last_scan_id != last_full_scan_id:
@@ -193,10 +193,8 @@ def get_result(project, filter_false_positive=False, threshold=0):
                 )
             )
         except Exception as e:
-            print(
-                "Fail to get scan result for scan id: {id}".format(id=last_full_scan_id)
-            )
-            print("Exception: {error} ".format(error=e))
+            logger.error("Fail to get scan result for scan id: {id}".format(id=last_full_scan_id))
+            logger.error("Exception: {error} ".format(error=e))
             return None
 
         result_list = merge_results_by_similarity_id(result_list, last_full_scan_result)

CheckmarxPythonSDK/CxOne/KeycloakAPI/ClientsApi.py

@@ -86,8 +86,8 @@ def post_clients(
             Relative path: /{realm}/clients
         """
         url = f"{self.base_url}/{realm}/clients"
-        print(f"Post request to {url}")
-        print(f"Request body: {client_representation.to_dict()}")
+        logger.info(f"Post request to {url}")
+        logger.info(f"Request body: {client_representation.to_dict()}")
         response = self.api_client.call_api("POST", url, json=client_representation.to_dict())
         return response.status_code == 201
 

CheckmarxPythonSDK/CxOne/repoManagerAPI.py

@@ -236,7 +236,7 @@ def get_all_repo_orgs_for_a_scm_type(
             url=url,
             params=params,
         )
-        print(response.json())
+        logger.info(response.json())
         return RepoOrgs(orgs=[RepoOrg(**item) for item in response.json().get("orgs")])
 
     def create_installation_of_scm_on_org(

CheckmarxPythonSDK/CxOne/reportAPI.py

@@ -2,6 +2,8 @@
 from CheckmarxPythonSDK.api_client import ApiClient
 from CheckmarxPythonSDK.CxOne.config import construct_configuration
 from typing import List
+import logging
+logger = logging.getLogger(__name__)
 
 
 class ReportAPI(object):
@@ -74,10 +76,10 @@ def create_scan_report_v2(
             )
             status = response.json().get("status")
             if status == "completed":
-                print("Report has been generated successfully!")
+                logger.info("Report has been generated successfully!")
                 break
             else:
-                print("Generating report, please wait...")
+                logger.info("Generating report, please wait...")
                 time.sleep(2)
         return report_id
 
@@ -123,10 +125,10 @@ def create_scan_report(
             )
             status = response.json().get("status")
             if status == "completed":
-                print("Report has been generated successfully!")
+                logger.info("Report has been generated successfully!")
                 break
             else:
-                print("Generating report, please wait...")
+                logger.info("Generating report, please wait...")
                 time.sleep(2)
         return report_id
 
@@ -233,13 +235,13 @@ def get_sca_scan_report(self, export_id: str) -> dict:
                 response = self.api_client.call_api(
                     method="GET", url=download_url
                 )
-                print("Report has been generated successfully!")
+                logger.info("Report has been generated successfully!")
                 break
             if status == "Failed":
-                print(f"Error: {response.content}")
+                logger.error(f"Error: {response.content}")
                 break
             else:
-                print("Generating report, please wait...")
+                logger.info("Generating report, please wait...")
                 time.sleep(2)
         return response.json()
 

CheckmarxPythonSDK/CxOne/scansAPI.py

@@ -5,6 +5,8 @@
 from httpx import Response
 from deprecated import deprecated
 from CheckmarxPythonSDK.utilities.compat import NO_CONTENT, OK, CREATED
+import logging
+logger = logging.getLogger(__name__)
 
 from .dto import (
     ScanInput,
@@ -481,10 +483,8 @@ def scan_by_repo_url(
                     {"type": engine, "value": engine_configs[engine]}
                 )
             else:
-                print(
-                    f"Warning: Engine '{engine}' is not supported "
-                    f"and will be ignored."
-                )
+                logger.warning(f"Warning: Engine '{engine}' is not supported "
+                    f"and will be ignored.")
         return self.api_client.call_api(
             method="POST", url=self.base_url, json=scan_data
         )

CheckmarxPythonSDK/CxOne/uploadsAPI.py

@@ -5,6 +5,8 @@
 from CheckmarxPythonSDK.utilities.compat import OK, NO_CONTENT
 from typing import List
 from os.path import exists
+import logging
+logger = logging.getLogger(__name__)
 
 
 class UploadsAPI(object):
@@ -46,7 +48,7 @@ def upload_zip_content_for_scanning(
             is_successful (bool)
         """
         if not zip_file_path or not exists(zip_file_path):
-            print("zip file path: {} does not exist".format(zip_file_path))
+            logger.info("zip file path: {} does not exist".format(zip_file_path))
         file_name = os.path.basename(zip_file_path)
         response = self.api_client.call_api(
             method="PUT",

CheckmarxPythonSDK/CxPortalSoapApiSDK/CxAuditWebService.py

@@ -1,6 +1,8 @@
 from os.path import exists
 from CheckmarxPythonSDK.configuration import Configuration
 from CheckmarxPythonSDK.CxPortalSoapApiSDK.config import construct_configuration
+import logging
+logger = logging.getLogger(__name__)
 from .sudsClient import SudsClient
 
 
@@ -438,7 +440,7 @@ def keep_alive(self) -> dict:
 
     def import_queries(self, imported_file_path: str) -> dict:
         if not exists(imported_file_path):
-            print("Error, the imported file {} not exist".format(imported_file_path))
+            logger.error("Error, the imported file {} not exist".format(imported_file_path))
             return None
         with open(imported_file_path, "rb") as xml_file:
             imported_file = xml_file.read()

CheckmarxPythonSDK/CxPortalSoapApiSDK/CxPortalWebService.py

@@ -9,6 +9,8 @@
 from typing import List, Union
 from CheckmarxPythonSDK.configuration import Configuration
 from CheckmarxPythonSDK.CxPortalSoapApiSDK.config import construct_configuration
+import logging
+logger = logging.getLogger(__name__)
 from .sudsClient import SudsClient
 
 
@@ -1152,7 +1154,7 @@ def import_preset(self, imported_file_path: str) -> Union[dict, None]:
 
         """
         if not exists(imported_file_path):
-            print("Error, the imported file {} not exist".format(imported_file_path))
+            logger.error("Error, the imported file {} not exist".format(imported_file_path))
             return
 
         with open(imported_file_path, "rb") as xml_file:
@@ -1177,7 +1179,7 @@ def import_queries(self, imported_file_path: str) -> Union[dict, None]:
 
         """
         if not exists(imported_file_path):
-            print("Error, the imported file {} not exist".format(imported_file_path))
+            logger.error("Error, the imported file {} not exist".format(imported_file_path))
             return
 
         with open(imported_file_path, "rb") as xml_file:

CheckmarxPythonSDK/CxReporting/api.py

@@ -4,6 +4,8 @@
 import json
 from CheckmarxPythonSDK.utilities.compat import OK, CREATED
 from typing import Union
+import logging
+logger = logging.getLogger(__name__)
 from .dto import (
     CreateReportDTO,
 )
@@ -97,12 +99,12 @@ def get_report(self, report_request: CreateReportDTO) -> Union[bytes, None]:
         )
         while report_status.upper() != "FINISHED":
             if "FAIL" in report_status.upper():
-                print("Report generation failed!")
+                logger.error("Report generation failed!")
                 return None
             report_status = self.retrieve_the_status_of_a_specific_report(
                 report_id=report_id
             )
-            print("report status: {}".format(report_status))
+            logger.info("report status: {}".format(report_status))
             time.sleep(2)
 
         return self.retrieve_the_file_of_a_specific_report(report_id=report_id)

CheckmarxPythonSDK/CxRestAPISDK/ScansAPI.py

@@ -6,6 +6,8 @@
 from os.path import normpath, exists, abspath
 
 from CheckmarxPythonSDK.utilities.compat import OK, CREATED, NO_CONTENT, ACCEPTED
+import logging
+logger = logging.getLogger(__name__)
 from .sast.projects.dto import CxLink, CxProject, CxPreset
 from .sast.engines.dto import CxEngineServer, CxEngineConfiguration
 from .sast.scans.dto import (
@@ -1020,9 +1022,7 @@ def create_new_scan_with_settings(
         url = f"{self.base_url}/cxrestapi/sast/scanWithSettings"
         file_name = os.path.basename(zipped_source_file_path)
         if not exists(normpath(abspath(zipped_source_file_path))):
-            print(
-                "zipped_source_file_path not exist: {}".format(zipped_source_file_path)
-            )
+            logger.info("zipped_source_file_path not exist: {}".format(zipped_source_file_path))
             return None
         fields = {
             "projectId": str(project_id),

CheckmarxPythonSDK/api_client.py

@@ -5,6 +5,8 @@
 import time
 import typing
 from typing import Callable, Union
+import logging
+logger = logging.getLogger(__name__)
 from .configuration import Configuration
 from .rate_limiter import RateLimiter
 from .__version__ import __version__
@@ -166,9 +168,7 @@ def wrapper(self, *args, **kwargs):
                                         base_wait * (2**retries), max_wait
                                     )
 
-                                print(
-                                    f"Rate limited (429), waiting {backoff_time:.2f} seconds before retrying..."
-                                )
+                                logger.info(f"Rate limited (429), waiting {backoff_time:.2f} seconds before retrying...")
                                 time.sleep(backoff_time)
                                 retries += 1
                                 continue
@@ -178,7 +178,7 @@ def wrapper(self, *args, **kwargs):
                     return response
 
                 except Exception as e:
-                    print(f"error: {e}")
+                    logger.error(f"error: {e}")
                     raise
 
             if last_exception:

CheckmarxPythonSDK/rate_limiter.py

@@ -1,5 +1,7 @@
 import time
 import threading
+import logging
+logger = logging.getLogger(__name__)
 
 
 class TokenBucket:
@@ -51,9 +53,7 @@ def consume(self, tokens: int = 1, block: bool = True) -> bool:
                 # Calculate wait time inside the lock, then sleep outside it
                 tokens_needed = tokens - self.tokens
                 wait_time = tokens_needed / self.refill_rate
-                print(
-                    f"Rate limiting: waiting {wait_time:.2f} seconds for {tokens_needed} tokens..."
-                )
+                logger.info(f"Rate limiting: waiting {wait_time:.2f} seconds for {tokens_needed} tokens...")
 
             # Sleep outside the lock so other threads are not blocked
             time.sleep(wait_time)