sedi-relay/relay/config.py

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',
	'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 (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)