import os import sys import yaml from functools import cached_property from pathlib import Path from platform import system from .misc import AppBase, DotDict DEFAULTS = { 'general_listen': '0.0.0.0', 'general_port': 8080, 'general_host': 'relay.example.com', 'database_type': 'sqlite', 'database_min_connections': 0, 'database_max_connections': 10, 'sqlite_database': Path('relay.sqlite3'), 'postgres_database': 'activityrelay', 'postgres_hostname': None, 'postgres_port': None, 'postgres_username': None, 'postgres_password': None, 'mysql_database': 'activityrelay', 'mysql_hostname': None, 'mysql_port': None, 'mysql_username': None, 'mysql_password': None } CATEGORY_NAMES = [ 'general', 'database', 'sqlite', 'postgres', 'mysql' ] def get_config_dir(): cwd = Path.cwd().joinpath('config.yaml') plat = system() if cwd.exists(): return cwd elif plat == 'Linux': cfgpath = Path('~/.config/activityrelay/config.yaml').expanduser() if cfgpath.exists(): return cfgpath etcpath = Path('/etc/activityrelay/config.yaml') if etcpath.exists() and os.getuid() == etcpath.stat().st_uid: return etcpath elif plat == 'Windows': cfgpath = Path('~/AppData/Roaming/activityrelay/config.yaml').expanduer() if cfgpath.exists(): return cfgpath elif plat == 'Darwin': cfgpath = Path('~/Library/Application Support/activityaelay/config.yaml') return cwd class Config(AppBase, dict): def __init__(self, path=None): DotDict.__init__(self, DEFAULTS) if self.is_docker: path = Path('/data/config.yaml') elif not path: path = get_config_dir() else: path = Path(path).expanduser() self._path = path self.load() def __setitem__(self, key, value): if key in {'database', 'hostname', 'port', 'username', 'password'}: key = f'{self.dbtype}_{key}' if (self.is_docker and key in {'general_host', 'general_port'}) or value == '__DEFAULT__': value = DEFAULTS[key] elif key in {'general_port', 'database_min_connections', 'database_max_connections'}: value = int(value) elif key == 'sqlite_database': if not isinstance(value, Path): value = Path(value) dict.__setitem__(self, key, value) @property def dbconfig(self): config = { 'type': self['database_type'], 'min_conn': self['database_min_connections'], 'max_conn': self['database_max_connections'] } if self.dbtype == 'sqlite': if not self['sqlite_database'].is_absolute(): config['database'] = self.path.with_name(str(self['sqlite_database'])).resolve() else: config['database'] = self['sqlite_database'].resolve() else: for key, value in self.items(): cat, name = key.split('_', 1) if self.dbtype == cat: config[name] = value return config @cached_property def is_docker(self): return bool(os.getenv('DOCKER_RUNNING')) @property def path(self): return self._path ## General config @property def host(self): return self['general_host'] @property def listen(self): return self['general_listen'] @property def port(self): return self['general_port'] ## Database config @property def dbtype(self): return self['database_type'] ## AP URLs @property def actor(self): return f'https://{self.host}/actor' @property def inbox(self): return f'https://{self.host}/inbox' @property def keyid(self): return f'{self.actor}#main-key' def reset(self): self.clear() self.update(DEFAULTS) def load(self): options = {} try: options['Loader'] = yaml.FullLoader except AttributeError: pass try: with open(self.path) as fd: config = yaml.load(fd, **options) except FileNotFoundError: return False for key, value in DEFAULTS.items(): cat, name = key.split('_', 1) self[key] = config.get(cat, {}).get(name, DEFAULTS[key]) def save(self): config = {key: {} for key in CATEGORY_NAMES} for key, value in self.items(): cat, name = key.split('_', 1) if isinstance(value, Path): value = str(value) config[cat][name] = value with open(self.path, 'w') as fd: yaml.dump(config, fd, sort_keys=False)