cirro.config

  1import configparser
  2import os
  3import re
  4from pathlib import Path
  5from typing import NamedTuple, Dict, Optional, List
  6
  7import requests
  8from requests import RequestException
  9
 10from cirro.models.tenant import Tenant
 11
 12
 13class Constants:
 14    home = os.environ.get('CIRRO_HOME', '~/.cirro')
 15    config_path = Path(home, 'config.ini').expanduser()
 16    default_base_url = 'cirro.bio'
 17    default_max_retries = 10
 18
 19
 20class UserConfig(NamedTuple):
 21    auth_method: str
 22    auth_method_config: Dict  # This needs to match the init params of the auth method
 23    base_url: Optional[str]
 24    transfer_max_retries: Optional[int]
 25    enable_additional_checksum: Optional[bool]
 26
 27
 28def extract_base_url(base_url: str):
 29    # remove http(s):// if it's there
 30    base_url = re.compile(r'https?://').sub('', base_url).strip()
 31    # remove anything after the first slash if it's there
 32    base_url = base_url.split('/')[0]
 33    return base_url
 34
 35
 36def list_tenants() -> List[Tenant]:
 37    resp = requests.get(f'https://nexus.{Constants.default_base_url}/info')
 38    resp.raise_for_status()
 39    return resp.json()['tenants']
 40
 41
 42def save_user_config(user_config: UserConfig):
 43    original_user_config = load_user_config()
 44    ini_config = configparser.ConfigParser()
 45    ini_config['General'] = {
 46        'auth_method': user_config.auth_method,
 47        'base_url': user_config.base_url,
 48        'transfer_max_retries': Constants.default_max_retries
 49    }
 50    if original_user_config:
 51        ini_config['General']['transfer_max_retries'] = str(original_user_config.transfer_max_retries)
 52
 53    ini_config[user_config.auth_method] = user_config.auth_method_config
 54    Constants.config_path.parent.mkdir(exist_ok=True)
 55    with Constants.config_path.open('w') as configfile:
 56        ini_config.write(configfile)
 57
 58
 59def load_user_config() -> Optional[UserConfig]:
 60    if not Constants.config_path.exists():
 61        return None
 62
 63    try:
 64        ini_config = configparser.ConfigParser()
 65        ini_config.read(str(Constants.config_path.absolute()))
 66        main_config = ini_config['General']
 67        auth_method = main_config.get('auth_method')
 68        base_url = main_config.get('base_url')
 69        transfer_max_retries = main_config.getint('transfer_max_retries', Constants.default_max_retries)
 70        enable_additional_checksum = main_config.getboolean('enable_additional_checksum', False)
 71
 72        if auth_method and ini_config.has_section(auth_method):
 73            auth_method_config = dict(ini_config[auth_method])
 74        else:
 75            auth_method_config = {}
 76
 77        return UserConfig(
 78            auth_method=auth_method,
 79            auth_method_config=auth_method_config,
 80            base_url=base_url,
 81            transfer_max_retries=transfer_max_retries,
 82            enable_additional_checksum=enable_additional_checksum
 83        )
 84    except Exception:
 85        raise RuntimeError('Configuration load error, please re-run configuration')
 86
 87
 88class AppConfig:
 89    def __init__(self, base_url: str = None):
 90        self.user_config = load_user_config()
 91        self.base_url = (base_url or
 92                         os.environ.get('CIRRO_BASE_URL') or
 93                         (self.user_config.base_url if self.user_config else None))
 94        if not self.base_url:
 95            raise RuntimeError('No base URL provided, please run cirro configure,'
 96                               ' set the CIRRO_BASE_URL environment variable, or provide the base_url parameter.')
 97        # remove http(s):// if it's there
 98        self.base_url = re.compile(r'https?://').sub('', self.base_url).strip()
 99        self.transfer_max_retries = self.user_config.transfer_max_retries\
100            if self.user_config else Constants.default_max_retries
101        self.enable_additional_checksum = self.user_config.enable_additional_checksum\
102            if self.user_config else False
103        self._init_config()
104
105    def _init_config(self):
106        self.rest_endpoint = f'https://{self.base_url}/api'
107        self.auth_endpoint = f'https://{self.base_url}/api/auth'
108
109        try:
110            info_resp = requests.get(f'{self.rest_endpoint}/info/system')
111            info_resp.raise_for_status()
112
113            info = info_resp.json()
114            self.client_id = info['auth']['sdkAppId']
115            self.user_pool_id = info['auth']['userPoolId']
116            self.references_bucket = info['referencesBucket']
117            self.resources_bucket = info['resourcesBucket']
118            self.region = info['region']
119        except RequestException:
120            raise RuntimeError(f'Failed connecting to {self.base_url}, please check your configuration')