cirro.config

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