fix: Refactor settings

Author: arajkumar

src/settings.py

@@ -5,19 +5,21 @@
 from pydantic import BaseSettings, HttpUrl
 
 
-# (fixme) Move all settings to read from here.
-class Settings(BaseSettings):
+class AggregatorSettings(BaseSettings):
     """Create Settings from env."""
 
-    snyk_package_url_format: HttpUrl = 'https://snyk.io/vuln/{ecosystem}:{package}'
-    snyk_signin_url: HttpUrl = 'https://snyk.io/login'
-    snyk_ecosystem_map: Dict[str, str] = {'pypi': 'pip'}
+    snyk_package_url_format: HttpUrl = "https://snyk.io/vuln/{ecosystem}:{package}"
+    snyk_signin_url: HttpUrl = "https://snyk.io/login"
+    snyk_ecosystem_map: Dict[str, str] = {"pypi": "pip"}
     disable_unknown_package_flow: bool = False
-    chester_service_host: str = 'not_set'
-    pypi_service_host: str = 'not_set'
-    golang_service_host: str = 'not_set'
-    hpf_service_host: str = 'not_set'
-    maven_service_host: str = 'not_set'
+
+
+class RecommenderSettings(BaseSettings):
+    """Create Recommender Settings from env."""
+
+    chester_service_host: str = "notset"
+    pypi_service_host: str = "notset"
+    maven_service_host: str = "notset"
     service_port: int = 6006
     unknown_packages_threshold: float = 0.3
     max_companion_packages: int = 5

src/stack_aggregator.py

@@ -12,7 +12,7 @@
 import requests
 import copy
 from collections import defaultdict
-from src.settings import Settings
+from src.settings import AggregatorSettings
 from src.utils import (select_latest_version, server_create_analysis, LICENSE_SCORING_URL_REST,
                        post_http_request, GREMLIN_SERVER_URL_REST, persist_data_in_db,
                        GREMLIN_QUERY_SIZE, format_date)
@@ -663,7 +663,7 @@ def execute(aggregated=None, persist=True):
                          'result': stack_data}
         # Ingestion of Unknown dependencies
         logger.info("Unknown ingestion flow process initiated.")
-        if Settings().disable_unknown_package_flow:
+        if AggregatorSettings().disable_unknown_package_flow:
             logger.warning('Skipping unknown flow %s', unknown_dep_list)
         else:
             try:

src/v2/recommender.py

@@ -14,7 +14,7 @@
 from src.utils import get_session_retry, persist_data_in_db, post_gremlin
 from src.v2.stack_aggregator import get_github_details, get_snyk_package_link
 from src.v2.normalized_packages import NormalizedPackages
-from src.settings import Settings
+from src.settings import RecommenderSettings
 
 from src.v2.models import (
     RecommenderRequest,
@@ -32,8 +32,10 @@ class InsightsRequest(BaseModel):
     ecosystem: Ecosystem
     transitive_stack: List[str] = []
     package_list: List[str]
-    unknown_packages_ratio_threshold: float = Settings().unknown_packages_threshold
-    comp_package_count_threshold: int = Settings().max_companion_packages
+    unknown_packages_ratio_threshold: float = (
+        RecommenderSettings().unknown_packages_threshold
+    )
+    comp_package_count_threshold: int = RecommenderSettings().max_companion_packages
 
 
 class InsightsCallException(Exception):
@@ -46,15 +48,17 @@ class InsightsWithEmptyPackageException(Exception):
 
 def _prepare_insights_url(host: str) -> str:
     assert host
-    url = "http://{host}:{port}".format(host=host, port=Settings().service_port)
+    url = "http://{host}:{port}".format(
+        host=host, port=RecommenderSettings().service_port
+    )
     endpoint = "{url}/api/v1/companion_recommendation".format(url=url)
     return endpoint
 
 
 ECOSYSTEM_TO_INSIGHTS_URL = {
-    "pypi": _prepare_insights_url(Settings().pypi_service_host),
-    "npm": _prepare_insights_url(Settings().chester_service_host),
-    "maven": _prepare_insights_url(Settings().maven_service_host),
+    "pypi": _prepare_insights_url(RecommenderSettings().pypi_service_host),
+    "npm": _prepare_insights_url(RecommenderSettings().chester_service_host),
+    "maven": _prepare_insights_url(RecommenderSettings().maven_service_host),
 }
 
 

src/v2/stack_aggregator.py

@@ -14,7 +14,7 @@
 from typing import Dict, List, Tuple, Set
 from f8a_utils.gh_utils import GithubUtils
 
-from src.settings import Settings
+from src.settings import AggregatorSettings
 from src.utils import (select_latest_version, server_create_analysis,
                        persist_data_in_db, post_gremlin, GREMLIN_QUERY_SIZE,
                        format_date)
@@ -171,8 +171,8 @@ def _has_vulnerability(pkg: PackageDetails) -> bool:
 # (fixme) link to snyk package should be identified during ingestion.
 def get_snyk_package_link(ecosystem: str, package: str) -> str:
     """Prepare snyk package link based on ecosystem and package name."""
-    ecosystem = Settings().snyk_ecosystem_map.get(ecosystem, ecosystem)
-    return Settings().snyk_package_url_format.format(ecosystem=ecosystem,
+    ecosystem = AggregatorSettings().snyk_ecosystem_map.get(ecosystem, ecosystem)
+    return AggregatorSettings().snyk_package_url_format.format(ecosystem=ecosystem,
                                                      package=quote(package, safe=''))
 
 
@@ -302,11 +302,11 @@ def get_result(self) -> StackAggregatorResult:
                                      analyzed_dependencies=package_details,
                                      unknown_dependencies=unknown_dependencies,
                                      license_analysis=license_analysis,
-                                     registration_link=Settings().snyk_signin_url)
+                                     registration_link=AggregatorSettings().snyk_signin_url)
 
     def initiate_unknown_package_ingestion(self):
         """Ingestion of Unknown dependencies."""
-        if Settings().disable_unknown_package_flow:
+        if AggregatorSettings().disable_unknown_package_flow:
             logger.warning('Skipping unknown flow %s', self.get_all_unknown_packages())
             return
 
@@ -383,7 +383,7 @@ def __init__(self, request: StackAggregatorRequest = None,
 
     def initiate_unknown_package_ingestion(self):
         """Ingestion of Unknown dependencies."""
-        if Settings().disable_unknown_package_flow:
+        if AggregatorSettings().disable_unknown_package_flow:
             logger.warning('Skipping unknown flow %s', self.get_all_unknown_packages())
             return
         logger.error('Ingestion is Not active for Golang.')