Source code for cityiq.config
# -*- coding: utf-8 -*-
# Copyright (c) 2019 Civic Knowledge. This file is licensed under the terms of the
# MIT License, included in this distribution as LICENSE
"""
"""
from os import environ
from pathlib import Path
from .exceptions import ConfigurationError
import yaml
[docs]class Config(object):
def __init__(self, path=None, **kwargs):
"""
:param path: A path in which to look for config. Prepended to search paths
kwargs set top level value.
Special handling for cache keys; kwargs of the
form "cache_object" will set the 'object' key of the top level
'cache' dict. So to change the object cache: Config(cache_object="/tmp")
Equivalently, the values in the config file in the 'cache" dict are flattened,
so 'cache->meta' is translated to 'cache_meta'
"""
self.parameters = 'client_id secret bbox zone uaa_url metadata_url event_url _config_file ' \
'start_time timezone cache_meta cache_objects cache_errors'.split()
self.env_vars = {e: f"CITYIQ_{e.upper()}" for e in self.parameters}
if not path:
self._paths = [
Path.cwd().joinpath('.city-iq.yaml'),
Path.cwd().joinpath('city-iq.yaml'),
Path.home().joinpath('.city-iq.yaml'),
]
else:
if Path(path).is_dir():
self._paths = [ Path(path).joinpath(e) for e in ['.city-iq.yaml', 'city-iq.yaml']]
else:
self._paths = [Path(path)]
if environ.get('CITYIQ_CONFIG'):
self._paths = [Path(environ.get('CITYIQ_CONFIG'))] + self._paths
self._kwargs = kwargs
self._config = self._load_config()
if 'cache' in self._config:
for k, v in self._config['cache'].items():
self._config['cache_'+k] = v
del self._config['cache']
def _load_config(self):
"""
Load a YAML configuration from the first configuration file that is found to exist
:return:
"""
for p in self._paths:
if p.exists():
with p.open() as f:
c = yaml.safe_load(f)
if c:
c['_config_file'] = str(p)
return c
else:
raise ConfigurationError(f"Didn't find a config file in paths: {self._paths}")
return {}
@property
def which(self):
return self._config_file
@property
def dict(self):
d = {k: self[k] for k in self.parameters}
for k, v in self._config.items():
if k not in d:
d[k] = v
return d
def __getattr__(self, item):
try:
return self._kwargs[item]
except KeyError:
pass
try:
return environ[self.env_vars[item]]
except KeyError:
pass
try:
return self._config[item]
except KeyError:
pass
if item not in self.parameters:
raise AttributeError(item)
return None
def __getitem__(self, item):
try:
return self.__getattr__(item)
except AttributeError:
raise IndexError(item)
def __str__(self):
import yaml
return yaml.dump(self.dict, default_flow_style=False)