[CveXplore-224] refactor

CveXplore/VERSION

@@ -1 +1 @@
-0.3.20.dev16
+0.3.20.dev17

CveXplore/common/data_source_connection.py

@@ -15,18 +15,39 @@ class DatasourceConnection(CveXploreObject):
  objects and generic database functions
  """
 
- # hack for documentation building
- if json.loads(os.getenv("DOC_BUILD"))["DOC_BUILD"] != "YES":
- __DATA_SOURCE_CONNECTION = DatabaseConnection(
- database_type="dummy",
- database_init_parameters={},
- ).database_connection
+ def __init__(self, collection: str):
+ """
+ Create a DatasourceConnection object
+ """
+ super().__init__()
+ self._collection = collection
+
+ @property
+ def datasource_connection(self):
+ # hack for documentation building
+ if json.loads(os.getenv("DOC_BUILD"))["DOC_BUILD"] != "YES":
+ return DatabaseConnection(
+ database_type="dummy",
+ database_init_parameters={},
+ ).database_connection
+ else:
+ return DatabaseConnection(
+ database_type=self.config.DATASOURCE_TYPE,
+ database_init_parameters=self.config.DATASOURCE_CONNECTION_DETAILS,
+ ).database_connection
+
+ @property
+ def datasource_collection_connection(self):
+ return getattr(self.datasource_connection, f"store_{self.collection}")
+
+ @property
+ def collection(self):
+ return self._collection
 
  def to_dict(self, *print_keys: str) -> dict:
  """
  Method to convert the entire object to a dictionary
  """
-
  if len(print_keys) != 0:
  full_dict = {
  k: v
@@ -40,30 +61,8 @@ def to_dict(self, *print_keys: str) -> dict:
 
  return full_dict
 
- def __init__(self, collection: str):
- """
- Create a DatasourceConnection object
- """
- super().__init__()
- self.__collection = collection
-
  def __eq__(self, other):
  return self.__dict__ == other.__dict__
 
  def __ne__(self, other):
  return self.__dict__ != other.__dict__
-
- @property
- def _datasource_connection(self):
- return DatasourceConnection.__DATA_SOURCE_CONNECTION
-
- @property
- def _datasource_collection_connection(self):
- return getattr(
- DatasourceConnection.__DATA_SOURCE_CONNECTION,
- f"store_{self.__collection}",
- )
-
- @property
- def _collection(self):
- return self.__collection

CveXplore/database/connection/dummy/dummy.py

@@ -10,3 +10,6 @@ def __init__(self, **kwargs):
  @property
  def dbclient(self):
  return self._dbclient
+
+ def set_handlers_for_collections(self):
+ pass

CveXplore/database/helpers/generic_db.py

@@ -56,15 +56,13 @@ def __init__(self, collection: str):
  }
 
  total_fields_list = (
- self.__default_fields + self.__fields_mapping[self._collection]
+ self.__default_fields + self.__fields_mapping[self.collection]
  )
  for field in total_fields_list:
  setattr(
  self,
  field,
- GenericDatabaseFieldsFunctions(
- field=field, collection=self._collection
- ),
+ GenericDatabaseFieldsFunctions(field=field, collection=self.collection),
  )
 
  def get_by_id(self, doc_id: str):
@@ -78,7 +76,7 @@ def get_by_id(self, doc_id: str):
  except ValueError:
  return "Provided value is not a string nor can it be cast to one"
 
- return self._datasource_collection_connection.find_one({"id": doc_id})
+ return self.datasource_collection_connection.find_one({"id": doc_id})
 
  def mget_by_id(self, *doc_ids: str) -> Union[Iterable[CveXploreObject], Iterable]:
  """
@@ -106,7 +104,7 @@ def _field_list(self, doc_id: str) -> list:
  map(
  lambda d: d.to_dict(),
  [
- self._datasource_collection_connection.find_one(
+ self.datasource_collection_connection.find_one(
  {"id": doc_id}
  )
  ],
@@ -139,7 +137,7 @@ def mapped_fields(self, collection: str) -> list:
 
  def __repr__(self):
  """String representation of object"""
- return f"<< {self.__class__.__name__}:{self._collection} >>"
+ return f"<< {self.__class__.__name__}:{self.collection} >>"
 
 
 class GenericDatabaseFieldsFunctions(DatasourceConnection):
@@ -164,7 +162,7 @@ def search(self, value: str):
 
  query = {self.__field: {"$regex": regex}}
 
- return self._datasource_collection_connection.find(query)
+ return self.datasource_collection_connection.find(query)
 
  def find(self, value: str | dict = None):
  """
@@ -176,8 +174,8 @@ def find(self, value: str | dict = None):
  else:
  query = None
 
- return self._datasource_collection_connection.find(query)
+ return self.datasource_collection_connection.find(query)
 
  def __repr__(self):
  """String representation of object"""
- return f"<< GenericDatabaseFieldsFunctions:{self._collection} >>"
+ return f"<< GenericDatabaseFieldsFunctions:{self.collection} >>"

CveXplore/database/helpers/specific_db.py

@@ -30,7 +30,7 @@ def get_cves_for_vendor(
  """
 
  the_result = list(
- self._datasource_collection_connection.find({"vendors": vendor})
+ self.datasource_collection_connection.find({"vendors": vendor})
  .limit(limit)
  .sort("cvss", DESCENDING)
  )
@@ -69,7 +69,7 @@ def get_by_id(self, doc_id: str):
  except ValueError:
  return "Provided value is not a string nor can it be cast to one"
 
- return self._datasource_collection_connection.find_one({"id": doc_id})
+ return self.datasource_collection_connection.find_one({"id": doc_id})
 
  def _field_list(self, doc_id: str) -> list:
  """
@@ -84,7 +84,7 @@ def _field_list(self, doc_id: str) -> list:
  map(
  lambda d: d.to_dict(),
  [
- self._datasource_collection_connection.find_one(
+ self.datasource_collection_connection.find_one(
  {"id": doc_id}
  )
  ],
@@ -116,7 +116,7 @@ def search_active_cpes(
  query = {"$and": [{field: {"$regex": regex}}, {"deprecated": False}]}
 
  the_result = list(
- self._datasource_collection_connection.find(query)
+ self.datasource_collection_connection.find(query)
  .limit(limit)
  .sort(field, sorting)
  )
@@ -136,7 +136,7 @@ def find_active_cpes(
  query = {"$and": [{field: value}, {"deprecated": False}]}
 
  the_result = list(
- self._datasource_collection_connection.find(query)
+ self.datasource_collection_connection.find(query)
  .limit(limit)
  .sort(field, sorting)
  )

CveXplore/objects/capec.py

@@ -26,7 +26,7 @@ def iter_related_weaknessess(self):
  if hasattr(self, "related_weakness"):
  if len(self.related_weakness) != 0:
  for each in self.related_weakness:
- cwe_doc = self._datasource_connection.store_cwe.find_one(
+ cwe_doc = self.datasource_connection.store_cwe.find_one(
  {"id": each}
  )
 
@@ -42,7 +42,7 @@ def iter_related_capecs(self):
  if hasattr(self, "related_capecs"):
  if len(self.related_capecs) != 0:
  for each in self.related_capecs:
- capec_doc = self._datasource_connection.store_capec.find_one(
+ capec_doc = self.datasource_connection.store_capec.find_one(
  {"id": each}
  )
 

CveXplore/objects/cpe.py

@@ -32,7 +32,7 @@ def iter_cves_matching_cpe(self, vuln_prod_search: bool = False):
 
  cpe_regex_string = create_cpe_regex_string(self.cpeName)
 
- results = self._datasource_connection.store_cves.find(
+ results = self.datasource_connection.store_cves.find(
  {cpe_searchField: {"$regex": cpe_regex_string}}
  ).sort("cvss", DESCENDING)
 

CveXplore/objects/cves.py

@@ -26,20 +26,20 @@ def __init__(self, **kwargs):
  try:
  if int(cwe_id):
  results = getattr(
- self._datasource_connection, "store_cwe"
+ self.datasource_connection, "store_cwe"
  ).find_one({"id": cwe_id})
  if results is not None:
  self.cwe = results
  except ValueError:
  pass
 
- capecs = self._datasource_connection.store_capec.find(
+ capecs = self.datasource_connection.store_capec.find(
  {"related_weakness": {"$in": [cwe_id]}}
  )
 
  setattr(self, "capec", list(capecs))
 
- via4s = self._datasource_connection.store_via4.find_one({"id": self.id})
+ via4s = self.datasource_connection.store_via4.find_one({"id": self.id})
 
  if via4s is not None:
  setattr(self, "via4_references", via4s)

CveXplore/objects/cvexplore_object.py

@@ -2,6 +2,7 @@
 CveXploreObject
 ===============
 """
+from CveXplore.common.config import Configuration
 
 
 class CveXploreObject(object):
@@ -10,7 +11,7 @@ class CveXploreObject(object):
  """
 
  def __init__(self):
- pass
+ self.config = Configuration
 
  def __repr__(self) -> str:
  return f"<< {self.__class__.__name__} >>"

CveXplore/objects/cwe.py

@@ -26,7 +26,7 @@ def iter_related_weaknessess(self):
  if hasattr(self, "related_weaknesses"):
  if len(self.related_weaknesses) != 0:
  for each in self.related_weaknesses:
- cwe_doc = self._datasource_connection.store_cwe.find_one(
+ cwe_doc = self.datasource_connection.store_cwe.find_one(
  {"id": each}
  )
 
@@ -40,7 +40,7 @@ def iter_related_capecs(self):
  :rtype: Capec
  """
 
- related_capecs = self._datasource_connection.store_capec.find(
+ related_capecs = self.datasource_connection.store_capec.find(
  {"related_weakness": self.id}
  )