Page MenuHome
No OneTemporary

"""Writes configuration to a config file in the home directory."""
import collections
import configparser
import datetime
import pathlib
import logging
from . import worker
HOME_CONFIG_FILE = pathlib.Path('~/.flamenco-worker.cfg').expanduser()
GLOBAL_CONFIG_FILE = pathlib.Path('./flamenco-worker.cfg').absolute()
CONFIG_SECTION = 'flamenco-worker'
'flamenco-worker': collections.OrderedDict([
('manager_url', ''),
('task_types', 'unknown sleep blender-render'),
('task_update_queue_db', 'flamenco-worker.db'),
('may_i_run_interval_seconds', '5'),
('worker_id', ''),
('worker_secret', ''),
# All intervals in seconds
('push_log_max_interval_seconds', str(worker.PUSH_LOG_MAX_INTERVAL.total_seconds())),
('push_log_max_entries', str(worker.PUSH_LOG_MAX_ENTRIES)),
('push_act_max_interval_seconds', str(worker.PUSH_ACT_MAX_INTERVAL.total_seconds())),
log = logging.getLogger(__name__)
class ConfigParser(configparser.ConfigParser):
"""ConfigParser that can easily get values from our default config section."""
_DEFAULT_INTERPOLATION = configparser.ExtendedInterpolation()
def value(self, key, valtype: type=str):
return valtype(self.get(CONFIG_SECTION, key))
def setvalue(self, key, value):
self.set(CONFIG_SECTION, key, value)
def interval_secs(self, key) -> datetime.timedelta:
"""Returns the configuration value as timedelta."""
secs = self.value(key, float)
return datetime.timedelta(seconds=secs)
def erase(self, key: str) -> bool:
return self.set(CONFIG_SECTION, key, '')
def merge_with_home_config(new_conf: dict):
"""Updates the home configuration file with the given config dict."""
confparser = ConfigParser()
confparser.read_dict({CONFIG_SECTION: {}}), encoding='utf8')
for key, value in new_conf.items():
confparser.set(CONFIG_SECTION, key, value)
tmpname = HOME_CONFIG_FILE.with_name( + '~')
log.debug('Writing configuration file to %s', tmpname)
with'wt', encoding='utf8') as outfile:
log.debug('Moving configuration file to %s', HOME_CONFIG_FILE)
tmpname.replace(HOME_CONFIG_FILE)'Updated configuration file %s', HOME_CONFIG_FILE)
def load_config(config_file: pathlib.Path = None,
show_effective_config: bool = False) -> ConfigParser:
"""Loads one or more configuration files."""
# Logging and the default interpolation of configparser both use the
# same syntax for variables. To make it easier to work with, we use
# another interpolation for config files, so they now use ${loglevel}
# whereas logging still uses %(levelname)s.
confparser = ConfigParser()
if config_file:'Loading configuration from %s', config_file)
if not config_file.exists():
log.fatal('Config file %s does not exist', config_file)
raise SystemExit()
loaded =, encoding='utf8')
if not GLOBAL_CONFIG_FILE.exists():
log.fatal('Config file %s does not exist', GLOBAL_CONFIG_FILE)
raise SystemExit()
filenames = [str(f.absolute()) for f in config_files]'Loading configuration from %s', ', '.join(filenames))
loaded =, encoding='utf8')'Succesfully loaded: %s', loaded)
if show_effective_config:
import sys'Effective configuration:')
to_show = configparser.ConfigParser(
if to_show.get(CONFIG_SECTION, 'worker_secret'):
to_show.set(CONFIG_SECTION, 'worker_secret', '-hidden-')
return confparser
def configure_logging(confparser: configparser.ConfigParser):
import logging.config
logging.config.fileConfig(confparser, disable_existing_loggers=True)

File Metadata

Mime Type
Mon, Aug 10, 8:58 PM (1 d, 23 h)
Storage Engine
Storage Format
Raw Data
Storage Handle

Event Timeline