Format Python code with psf/black (#190)

Co-authored-by: P-T-I <[email protected]>

CveXplore/api/nvd_nist/nvd_nist_api.py

@@ -63,11 +63,9 @@ def __init__(
  self.max_page_length = namedtuple("max_page_length", "CVE CPE")(2000, 10000)
 
  def get_url_only(self, resource: dict = None, data: int = 1) -> str:
-
  return self._build_url(resource=resource, data=data)
 
  def _build_url(self, resource: dict = None, data: int = 1) -> str:
-
  if resource is not None:
  resource = urlencode(resource)
  if data == self.datasource.CVE:
@@ -95,7 +93,6 @@ def _connect(
  timeout: int = 60,
  return_response_object: bool = False,
  ):
-
  requests.packages.urllib3.disable_warnings()
 
  request_api_resource = {
@@ -138,7 +135,6 @@ def __repr__(self):
  return f"<< NvdNistApi:{self.baseurl} >>"
 
  def get_cves_from_start_year(self):
-
  start_date = datetime(int(self.config.CVE_START_YEAR), 1, 1, 0, 0, 0, 0)
  start_date_iso = start_date.isoformat()
 
@@ -158,7 +154,6 @@ def check_date_range(
  last_mod_start_date: datetime = None,
  last_mod_end_date: datetime = None,
  ):
-
  # Check if diff > 120 days
  diff = last_mod_end_date - last_mod_start_date
 
@@ -203,7 +198,6 @@ def get_all_data(
  last_mod_start_date: datetime = None,
  last_mod_end_date: datetime = None,
  ):
-
  resource = {}
 
  if last_mod_start_date is not None and last_mod_end_date is not None:
@@ -223,7 +217,6 @@ def get_all_data(
  )
 
  if isinstance(data, int):
-
  for each_data in ApiData(
  results_per_page=getattr(self.max_page_length, data_type.upper()),
  start_index=0,
@@ -310,12 +303,10 @@ def __iter__(self):
  return self
 
  def __next__(self):
-
  if (
  self._current_index < self._total_results
  or self._page_length == self._total_results
  ):
-
  start_time = time.time()
 
  if not self.last_stop_time == 0:
@@ -368,7 +359,6 @@ def __next__(self):
  raise StopIteration
 
  def process_async(self):
-
  loop = asyncio.new_event_loop()
  asyncio.set_event_loop(loop)
  results = loop.run_until_complete(self.fetch_all(loop))

CveXplore/cli_cmds/cpe_cmds/commands.py

@@ -103,7 +103,6 @@ def search_cmd(
  pretty,
  output,
 ):
-
  if not name and not vendor and title:
  search_by = "title"
  elif name and not vendor and title:

CveXplore/cli_cmds/cve_cmds/commands.py

@@ -54,7 +54,6 @@ def search_cmd(
  pretty,
  output,
 ):
-
  if cve:
  ret_list = ctx.obj["data_source"].cves_by_id(*cve)
  else:

CveXplore/cli_cmds/stats_cmds/commands.py

@@ -7,7 +7,6 @@
 @click.option("--pretty", is_flag=True, help="Pretty print the output")
 @click.pass_context
 def stats_cmd(ctx, pretty, output="json"):
-
  if ctx.invoked_subcommand is None:
  printer(
  input_data=ctx.obj["data_source"].get_db_content_stats(),

CveXplore/database/connection/mongo_db.py

@@ -54,7 +54,6 @@ def __init__(
  atexit.register(self.disconnect)
 
  def get_collections_details(self):
-
  for each in self._dbclient.list_collections():
  yield each
 
@@ -69,7 +68,6 @@ def set_handlers_for_collections(self):
 
  @property
  def get_collection_names(self):
-
  return self._dbclient.list_collection_names()
 
  def disconnect(self):

CveXplore/database/maintenance/DatabaseSchemaChecker.py

@@ -49,7 +49,6 @@ def create_indexes(self):
 
  def update(self):
  try:
-
  current_record = list(self.dbh.find({}))
 
  if len(current_record) != 0:

CveXplore/database/maintenance/Sources_process.py

@@ -117,9 +117,7 @@ def process_downloads(self, sites: list = None):
  )
 
  if self.do_process:
-
  if not self.is_update:
-
  total_results = self.api_handler.get_count(
  self.api_handler.datasource.CPE
  )
@@ -756,9 +754,7 @@ def process_downloads(self, sites: list = None):
  )
 
  if self.do_process:
-
  if not self.is_update:
-
  total_results = self.api_handler.get_count(
  self.api_handler.datasource.CVE
  )
@@ -859,7 +855,6 @@ def update(self):
  return self.last_modified
 
  def populate(self):
-
  self.logger.info("CVE database population started")
 
  self.logger.info(
@@ -900,7 +895,6 @@ def __init__(self):
  self.logger = logging.getLogger("VIADownloads")
 
  def file_to_queue(self, file_tuple: Tuple[str, str]):
-
  working_dir, filename = file_tuple
 
  for cve in self.ijson_handler.fetch(filename=filename, prefix=self.prefix):
@@ -930,7 +924,6 @@ def file_to_queue(self, file_tuple: Tuple[str, str]):
  )
 
  def process_item(self, item: dict):
-
  if self.is_update:
  self.queue.put(
  DatabaseAction(
@@ -991,7 +984,6 @@ def __init__(self):
  self.parser.setContentHandler(self.ch)
 
  def file_to_queue(self, file_tuple: Tuple[str, str]):
-
  working_dir, filename = file_tuple
 
  self.parser.parse(filename)
@@ -1053,7 +1045,6 @@ def __init__(self):
  self.parser.setContentHandler(self.ch)
 
  def file_to_queue(self, file_tuple: Tuple[str, str]):
-
  working_dir, filename = file_tuple
 
  for f in glob.glob(f"{working_dir}/*.xml"):
@@ -1113,7 +1104,6 @@ class DatabaseIndexer(object):
  """
 
  def __init__(self):
-
  database = MongoDBConnection(**json.loads(os.getenv("MONGODB_CON_DETAILS")))
  self.database = database._dbclient
 
@@ -1175,7 +1165,6 @@ def getInfo(self, collection: str):
  return sanitize(self.database["info"].find_one({"db": collection}))
 
  def create_indexes(self, collection: str = None):
-
  if collection is not None:
  try:
  for each in self.indexes[collection]:

CveXplore/database/maintenance/content_handlers.py

@@ -77,7 +77,6 @@ def __init__(self):
  self.execution_flow = defaultdict(dict)
 
  def startElement(self, name, attrs):
-
  if name == "Attack_Pattern_Catalog":
  self.Attack_Pattern_Catalog_tag = True
  if name == "Attack_Patterns" and self.Attack_Pattern_Catalog_tag:
@@ -230,7 +229,6 @@ def endElement(self, name):
  )
 
  elif self.taxonomy_name == "WASC":
-
  if "/" in self.entry_name_ch:
  url = "http://projects.webappsec.org/{}".format(
  self.entry_name_ch.replace("/", " and ").replace(" ", "-")
@@ -385,7 +383,6 @@ def __init__(self):
  self.category_relationships_tag = False
 
  def startElement(self, name, attrs):
-
  if name == "Weakness":
  self.weakness_tag = True
  self.statement = ""

CveXplore/database/maintenance/db_action.py

@@ -7,7 +7,6 @@ class DatabaseAction(object):
  actions = collections.namedtuple("Actions", "InsertOne UpdateOne")(0, 1)
 
  def __init__(self, action: actions, collection: str, doc: dict):
-
  self.action = action
  self.collection = collection
  self.doc = doc

CveXplore/database/maintenance/main_updater.py

@@ -62,7 +62,6 @@ def update(self, update_source: str | list = None):
 
  try:
  if update_source is None:
-
  for source in self.sources:
  up = source["updater"]()
  up.update()

CveXplore/database/maintenance/worker_q.py

@@ -5,7 +5,6 @@
 
 class WorkerQueue(Queue):
  def __init__(self, name: str, maxsize: int = 0):
-
  super().__init__(maxsize)
  self.name = name
  self.maxsize = maxsize

CveXplore/main.py

@@ -324,7 +324,6 @@ def get_db_content_stats(self) -> dict | str:
  stats = defaultdict(dict)
 
  if not isinstance(self.datasource, ApiDatabaseSource):
-
  if hasattr(self.datasource, "store_info"):
  results = self.datasource.store_info.find({})
  for each in results:

CveXplore/objects/capec.py

@@ -11,7 +11,6 @@ class Capec(DatasourceConnection):
  """
 
  def __init__(self, **kwargs):
-
  super().__init__("capec")
 
  for each in kwargs:

CveXplore/objects/cpe.py

@@ -16,7 +16,6 @@ class Cpe(DatasourceConnection):
  """
 
  def __init__(self, **kwargs):
-
  super().__init__("cpe")
 
  for each in kwargs:

CveXplore/objects/cves.py

@@ -13,7 +13,6 @@ class Cves(DatasourceConnection):
  """
 
  def __init__(self, **kwargs):
-
  super().__init__("cves")
 
  for each in kwargs:

CveXplore/objects/cwe.py

@@ -11,7 +11,6 @@ class Cwe(DatasourceConnection):
  """
 
  def __init__(self, **kwargs):
-
  super().__init__("cwe")
 
  for each in kwargs: