PILOT-3306: Refactor settings usage

.github/workflows/run-tests.yml

@@ -47,7 +47,7 @@ jobs:
         uses: pre-commit/action@v3.0.0
 
       - name: Run tests
-        run: poetry run pytest -vvv --exitfirst --cov=app --cov-report=term --cov-report=xml --cov-fail-under=54
+        run: poetry run pytest -vvv --exitfirst --cov=app --cov-report=term --cov-report=xml --cov-fail-under=60
 
       - name: Coverage report comment
         uses: mishakav/pytest-coverage-comment@v1.1.42

app/configs/app_config.py

@@ -2,16 +2,11 @@
 #
 # Contact Indoc Research for any questions regarding the use of this source code.
 
-from env import ConfigClass
+from app.configs.config import ConfigClass
 
 
 class AppConfig:
     class Env:
-        section = 'environment'
-        project = ConfigClass.project
-        user_config_path = ConfigClass.config_path
-        msg_path = ConfigClass.custom_path
-        user_config_file = 'config.ini'
         token_warn_need_refresh = 120  # refresh token if token is about to expire
         token_refresh_interval = 90  # auto refresh token every 40 seconds
 
@@ -23,21 +18,16 @@ class Env:
         resilient_backoff = 1
         resilient_retry_interval = 1  # seconds
         resilient_retry_code = [502, 503, 504, 404, 401]
-        pipeline_straight_upload = f'{project}cli_upload'
-        default_upload_message = f'{project}cli straight uploaded'
+        pipeline_straight_upload = f'{ConfigClass.project}cli_upload'
+        default_upload_message = f'{ConfigClass.project}cli straight uploaded'
         session_duration = 3600.0
         upload_batch_size = 100
-        harbor_client_secret = ConfigClass.harbor_client_secret
         core_zone = 'core'
         green_zone = 'greenroom'
         core_bucket_prefix = 'core'
         greenroom_bucket_prefix = 'gr'
 
-        keycloak_device_client_id = ConfigClass.keycloak_device_client_id
-        keycloak_api_key_audience = ConfigClass.keycloak_api_key_audience
-
     class Connections:
-        section = 'connections'
         url_harbor = ConfigClass.url_harbor
         url_authn = ConfigClass.url_authn
         url_refresh_token = ConfigClass.url_refresh_token
@@ -53,7 +43,5 @@ class Connections:
         url_validation = ConfigClass.url_validation
         url_keycloak = ConfigClass.url_keycloak
         url_keycloak_token = f'{ConfigClass.url_keycloak}/token'
-        url_keycloak_realm = ConfigClass.url_keycloak.rstrip('/').replace('/protocol/openid-connect', '')
         url_bff = ConfigClass.url_bff
-        # add url_base to check if value exist
         url_base = ConfigClass.base_url

app/configs/config.py

@@ -0,0 +1,107 @@
+# Copyright (C) 2022-2023 Indoc Research
+#
+# Contact Indoc Research for any questions regarding the use of this source code.
+
+from functools import lru_cache
+from pathlib import Path
+from typing import Set
+
+from pydantic import computed_field
+from pydantic_settings import BaseSettings
+from pydantic_settings import SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file='.env', extra='allow')
+
+    project: str = 'pilot'
+    app_name: str = 'pilotcli'
+
+    @computed_field
+    def config_path(self) -> str:
+        return str(Path.home() / f'.{self.app_name}')
+
+    config_file: str = 'config.ini'
+
+    keycloak_device_client_id: str = 'cli'
+    keycloak_api_key_audience: Set[str] = {'api-key'}
+
+    vm_info: str = ''
+
+    harbor_client_secret: str = ''
+    url_harbor: str = ''
+
+    domain: str = 'pilot.indocresearch.com'
+
+    @computed_field
+    def base_url(self) -> str:
+        return f'https://api.{self.domain}/pilot'
+
+    @computed_field
+    def url_bff(self) -> str:
+        return f'{self.base_url}/cli'
+
+    @computed_field
+    def url_keycloak_realm(self) -> str:
+        return f'https://iam.{self.domain}/realms/pilot'
+
+    @computed_field
+    def url_keycloak(self) -> str:
+        return f'{self.url_keycloak_realm}/protocol/openid-connect'
+
+    @computed_field
+    def url_authn(self) -> str:
+        return f'{self.base_url}/portal/users/auth'
+
+    @computed_field
+    def url_refresh_token(self) -> str:
+        return f'{self.base_url}/portal/users/refresh'
+
+    @computed_field
+    def url_file_tag(self) -> str:
+        return f'{self.base_url}/portal/v2/%s/tags'
+
+    @computed_field
+    def url_upload_greenroom(self) -> str:
+        return f'{self.base_url}/upload/gr'
+
+    @computed_field
+    def url_upload_core(self) -> str:
+        return f'{self.base_url}/upload/core'
+
+    @computed_field
+    def url_status(self) -> str:
+        return f'{self.base_url}/portal/v1/files/actions/tasks'
+
+    @computed_field
+    def url_download_greenroom(self) -> str:
+        return f'{self.base_url}/portal/download/gr/'
+
+    @computed_field
+    def url_download_core(self) -> str:
+        return f'{self.base_url}/portal/download/core/'
+
+    @computed_field
+    def url_v2_download_pre(self) -> str:
+        return f'{self.url_bff}/v1/project/%s/files/download'
+
+    @computed_field
+    def url_dataset_v2download(self) -> str:
+        return f'{self.base_url}/portal/download/core/v2/dataset'
+
+    @computed_field
+    def url_dataset(self) -> str:
+        return f'{self.base_url}/portal/v1/dataset'
+
+    @computed_field
+    def url_validation(self) -> str:
+        return f'{self.base_url}/v1/files/validation'
+
+
+@lru_cache(1)
+def get_settings() -> Settings:
+    settings = Settings()
+    return settings
+
+
+ConfigClass = get_settings()

app/configs/user_config.py

@@ -10,7 +10,7 @@
 from typing import Iterable
 from typing import Union
 
-from app.configs.app_config import AppConfig
+from app.configs.config import ConfigClass
 from app.models.singleton import Singleton
 from app.services.crypto.crypto import decryption
 from app.services.crypto.crypto import encryption
@@ -28,9 +28,9 @@ class UserConfig(metaclass=Singleton):
 
     def __init__(self, config_path: Union[str, Path, None] = None, config_filename: Union[str, None] = None) -> None:
         if config_path is None:
-            config_path = AppConfig.Env.user_config_path
+            config_path = ConfigClass.config_path
         if config_filename is None:
-            config_filename = AppConfig.Env.user_config_file
+            config_filename = ConfigClass.config_file
 
         config_path = Path(config_path)
         if not config_path.exists():

app/services/file_manager/file_upload/upload_client.py

@@ -18,6 +18,7 @@
 
 import app.services.output_manager.message_handler as mhandler
 from app.configs.app_config import AppConfig
+from app.configs.config import ConfigClass
 from app.configs.user_config import UserConfig
 from app.models.upload_form import generate_on_success_form
 from app.services.file_manager.file_upload.models import FileObject
@@ -413,7 +414,7 @@ def check_status(self, file_object: FileObject) -> bool:
     def set_finish_upload(self):
         self.finish_upload = True
 
-    def upload_token_refresh(self, azp: str = AppConfig.Env.keycloak_device_client_id):
+    def upload_token_refresh(self, azp: str = ConfigClass.keycloak_device_client_id):
         token_manager = SrvTokenManager()
         DEFAULT_INTERVAL = 2  # seconds to check if the upload is finished
         total_count = 0  # when total_count equals token_refresh_interval, refresh token

app/services/user_authentication/decorator.py

@@ -4,16 +4,15 @@
 
 from functools import wraps
 
-from app.configs.app_config import AppConfig
+from app.configs.config import ConfigClass
 from app.services.output_manager.error_handler import ECustomizedError
 from app.services.output_manager.error_handler import SrvErrorHandler
+from app.services.user_authentication.token_manager import SrvTokenManager
+from app.services.user_authentication.user_login_logout import check_is_active
+from app.services.user_authentication.user_login_logout import check_is_login
 
-from .token_manager import SrvTokenManager
-from .user_login_logout import check_is_active
-from .user_login_logout import check_is_login
 
-
-def require_valid_token(azp=AppConfig.Env.keycloak_device_client_id):
+def require_valid_token(azp=ConfigClass.keycloak_device_client_id):
     def decorate(func):
         @wraps(func)
         def decorated(*args, **kwargs):

app/services/user_authentication/token_manager.py

@@ -8,6 +8,7 @@
 import requests
 
 from app.configs.app_config import AppConfig
+from app.configs.config import ConfigClass
 from app.configs.user_config import UserConfig
 from app.models.enums import LoginMethod
 from app.models.service_meta_class import MetaService
@@ -44,7 +45,7 @@ def is_api_key(self) -> bool:
         audience = token['aud']
         if isinstance(audience, str):
             audience = [audience]
-        return AppConfig.Env.keycloak_api_key_audience.issubset(set(audience))
+        return ConfigClass.keycloak_api_key_audience.issubset(set(audience))
 
     def check_valid(self, required_azp):
         """
@@ -64,7 +65,7 @@ def check_valid(self, required_azp):
             # ``kong`` is hardcoded in the decorator definition as default value.
             azp_token_condition = decoded_access_token['azp'] not in [
                 required_azp,
-                AppConfig.Env.keycloak_device_client_id,
+                ConfigClass.keycloak_device_client_id,
             ]
 
             if azp_token_condition or expiry_at <= now:
@@ -86,7 +87,7 @@ def refresh(self, azp: str) -> None:
         }
 
         if azp == 'harbor':
-            payload.update({'client_id': AppConfig.Env.harbor_client_secret})
+            payload.update({'client_id': ConfigClass.harbor_client_secret})
 
         headers = {'Content-Type': 'application/x-www-form-urlencoded'}
         response = requests.post(url, data=payload, headers=headers)

app/services/user_authentication/user_login_logout.py

@@ -13,6 +13,7 @@
 from requests import RequestException
 
 from app.configs.app_config import AppConfig
+from app.configs.config import ConfigClass
 from app.configs.user_config import UserConfig
 from app.services.output_manager.error_handler import ECustomizedError
 from app.services.output_manager.error_handler import SrvErrorHandler
@@ -22,7 +23,7 @@
 def exchange_api_key(api_key: str) -> Union[str, None]:
     """Exchange API Key with JWT token using Keycloak."""
 
-    url = f'{AppConfig.Connections.url_keycloak_realm}/api-key/{api_key}'
+    url = f'{ConfigClass.url_keycloak_realm}/api-key/{api_key}'
     try:
         response = requests.get(url, timeout=5)
         response.raise_for_status()
@@ -59,7 +60,7 @@ def user_device_id_login() -> Dict[str, Any]:
 
     url = f'{AppConfig.Connections.url_keycloak}/auth/device'
     headers = {'Content-Type': 'application/x-www-form-urlencoded'}
-    data = {'client_id': AppConfig.Env.keycloak_device_client_id}
+    data = {'client_id': ConfigClass.keycloak_device_client_id}
     resp = requests.post(url, headers=headers, data=data)
     if resp.status_code == 200:
         device_data = resp.json()
@@ -80,7 +81,7 @@ def validate_user_device_login(device_code: str, expires: int, interval: int) ->
     headers = {'Content-Type': 'application/x-www-form-urlencoded'}
     data = {
         'device_code': device_code,
-        'client_id': AppConfig.Env.keycloak_device_client_id,
+        'client_id': ConfigClass.keycloak_device_client_id,
         'grant_type': 'urn:ietf:params:oauth:grant-type:device_code',
     }
     waiting_result = True
@@ -162,7 +163,7 @@ def request_harbor_tokens(username, password):
         'username': username,
         'password': password,
         'client_id': 'harbor',
-        'client_secret': AppConfig.Env.harbor_client_secret,
+        'client_secret': ConfigClass.harbor_client_secret,
     }
     headers = {'Content-Type': 'application/x-www-form-urlencoded'}
     response = requests.post(url, data=payload, headers=headers, verify=False)

app/utils/aggregated.py

@@ -10,16 +10,16 @@
 import requests
 
 from app.configs.app_config import AppConfig
+from app.configs.config import ConfigClass
 from app.configs.user_config import UserConfig
 from app.services.output_manager.error_handler import ECustomizedError
 from app.services.output_manager.error_handler import SrvErrorHandler
 from app.services.user_authentication.decorator import require_valid_token
-from env import ConfigClass
 
 
 def resilient_session():
     # each resilient session will
-    headers = {'VM-Info': ConfigClass.VM_INFO}
+    headers = {'VM-Info': ConfigClass.vm_info}
     client = httpx.Client(headers=headers, timeout=None)
     return client
 
@@ -39,6 +39,8 @@ def search_item(project_code, zone, folder_relative_path, item_type, container_t
     res = requests.get(url, params=params, headers=headers)
     if res.status_code == 403:
         SrvErrorHandler.customized_handle(ECustomizedError.PERMISSION_DENIED, project_code)
+    elif res.status_code == 404:
+        pass
     elif res.status_code != 200:
         SrvErrorHandler.default_handle(res.text, True)