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
 24
 25def extract_base_url(base_url: str):
 26    # remove http(s):// if it's there
 27    base_url = re.compile(r'https?://').sub('', base_url).strip()
 28    # remove anything after the first slash if it's there
 29    base_url = base_url.split('/')[0]
 30    return base_url
 31
 32
 33def save_user_config(user_config: UserConfig):
 34    original_user_config = load_user_config()
 35    ini_config = configparser.ConfigParser()
 36    ini_config['General'] = {
 37        'auth_method': user_config.auth_method,
 38        'base_url': user_config.base_url,
 39        'transfer_max_retries': Constants.default_max_retries
 40    }
 41    if original_user_config:
 42        ini_config['General']['transfer_max_retries'] = str(original_user_config.transfer_max_retries)
 43
 44    ini_config[user_config.auth_method] = user_config.auth_method_config
 45    Constants.config_path.parent.mkdir(exist_ok=True)
 46    with Constants.config_path.open('w') as configfile:
 47        ini_config.write(configfile)
 48
 49
 50def load_user_config() -> Optional[UserConfig]:
 51    if not Constants.config_path.exists():
 52        return None
 53
 54    try:
 55        ini_config = configparser.ConfigParser()
 56        ini_config.read(str(Constants.config_path.absolute()))
 57        main_config = ini_config['General']
 58        auth_method = main_config.get('auth_method')
 59        base_url = main_config.get('base_url')
 60        transfer_max_retries = main_config.getint('transfer_max_retries', Constants.default_max_retries)
 61
 62        if auth_method and ini_config.has_section(auth_method):
 63            auth_method_config = dict(ini_config[auth_method])
 64        else:
 65            auth_method_config = {}
 66
 67        return UserConfig(
 68            auth_method=auth_method,
 69            auth_method_config=auth_method_config,
 70            base_url=base_url,
 71            transfer_max_retries=transfer_max_retries
 72        )
 73    except Exception:
 74        raise RuntimeError('Configuration load error, please re-run configuration')
 75
 76
 77class AppConfig:
 78    def __init__(self, base_url: str = None):
 79        self.user_config = load_user_config()
 80        self.base_url = (base_url or
 81                         os.environ.get('CIRRO_BASE_URL') or
 82                         (self.user_config.base_url if self.user_config else None))
 83        if not self.base_url:
 84            raise RuntimeError('No base URL provided, please run cirro configure,'
 85                               ' set the CIRRO_BASE_URL environment variable, or provide the base_url parameter.')
 86        # remove http(s):// if it's there
 87        self.base_url = re.compile(r'https?://').sub('', self.base_url).strip()
 88        self.transfer_max_retries = self.user_config.transfer_max_retries\
 89            if self.user_config else Constants.default_max_retries
 90        self._init_config()
 91
 92    @property
 93    def checksum_method_display(self):
 94        return self.checksum_method.replace('NVME', '')
 95
 96    @property
 97    def checksum_method(self):
 98        return 'CRC64NVME'
 99
100    def _init_config(self):
101        self.rest_endpoint = f'https://{self.base_url}/api'
102        self.auth_endpoint = f'https://{self.base_url}/api/auth'
103
104        try:
105            info_resp = requests.get(f'{self.rest_endpoint}/info/system')
106            info_resp.raise_for_status()
107
108            info = info_resp.json()
109            self.client_id = info['auth']['sdkAppId']
110            self.user_pool_id = info['auth']['userPoolId']
111            self.references_bucket = info['referencesBucket']
112            self.resources_bucket = info['resourcesBucket']
113            self.region = info['region']
114        except RequestException:
115            raise RuntimeError(f'Failed connecting to {self.base_url}, please check your configuration')