+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
"""
Nominatim configuration accessor.
"""
import logging
import os
from pathlib import Path
+import json
import yaml
from dotenv import dotenv_values
from nominatim.errors import UsageError
LOG = logging.getLogger()
+CONFIG_CACHE = {}
+
+def flatten_config_list(content, section=''):
+ """ Flatten YAML configuration lists that contain include sections
+ which are lists themselves.
+ """
+ if not content:
+ return []
+
+ if not isinstance(content, list):
+ raise UsageError(f"List expected in section '{section}'.")
+
+ output = []
+ for ele in content:
+ if isinstance(ele, list):
+ output.extend(flatten_config_list(ele, section))
+ else:
+ output.append(ele)
+
+ return output
+
class Configuration:
""" Load and manage the project configuration.
if project_dir is not None and (project_dir / '.env').is_file():
self._config.update(dotenv_values(str((project_dir / '.env').resolve())))
- # Add defaults for variables that are left empty to set the default.
- # They may still be overwritten by environment variables.
- if not self._config['NOMINATIM_ADDRESS_LEVEL_CONFIG']:
- self._config['NOMINATIM_ADDRESS_LEVEL_CONFIG'] = \
- str(config_dir / 'address-levels.json')
-
class _LibDirs:
pass
def __getattr__(self, name):
name = 'NOMINATIM_' + name
- return self.environ.get(name) or self._config[name]
+ if name in self.environ:
+ return self.environ[name]
+
+ return self._config[name]
def get_bool(self, name):
""" Return the given configuration parameter as a boolean.
raise UsageError("Configuration error.") from exp
+ def get_path(self, name):
+ """ Return the given configuration parameter as a Path.
+ If a relative path is configured, then the function converts this
+ into an absolute path with the project directory as root path.
+ If the configuration is unset, a falsy value is returned.
+ """
+ value = self.__getattr__(name)
+ if value:
+ value = Path(value)
+
+ if not value.is_absolute():
+ value = self.project_dir / value
+
+ value = value.resolve()
+
+ return value
+
def get_libpq_dsn(self):
""" Get configured database DSN converted into the key/value format
understood by libpq and psycopg.
style = self.__getattr__('IMPORT_STYLE')
if style in ('admin', 'street', 'address', 'full', 'extratags'):
- return self.config_dir / 'import-{}.style'.format(style)
+ return self.config_dir / f'import-{style}.style'
- return Path(style)
+ return self.find_config_file('', 'IMPORT_STYLE')
def get_os_env(self):
is loaded using this function and added at the position in the
configuration tree.
"""
- configfile = self._find_config_file(filename, config)
+ configfile = self.find_config_file(filename, config)
- if configfile.suffix != '.yaml':
- LOG.format("Format error while reading '%s': only YAML format supported.",
- configfile)
- raise UsageError("Cannot handle config file format.")
+ if str(configfile) in CONFIG_CACHE:
+ return CONFIG_CACHE[str(configfile)]
+
+ if configfile.suffix in ('.yaml', '.yml'):
+ result = self._load_from_yaml(configfile)
+ elif configfile.suffix == '.json':
+ with configfile.open('r', encoding='utf-8') as cfg:
+ result = json.load(cfg)
+ else:
+ raise UsageError(f"Config file '{configfile}' has unknown format.")
- return self._load_from_yaml(configfile)
+ CONFIG_CACHE[str(configfile)] = result
+ return result
- def _find_config_file(self, filename, config=None):
+ def find_config_file(self, filename, config=None):
""" Resolve the location of a configuration file given a filename and
an optional configuration option with the file name.
Raises a UsageError when the file cannot be found or is not
if cfg_filename:
cfg_filename = Path(cfg_filename)
- if not cfg_filename.is_absolute():
- cfg_filename = self.project_dir / cfg_filename
+ if cfg_filename.is_absolute():
+ cfg_filename = cfg_filename.resolve()
- cfg_filename = cfg_filename.resolve()
+ if not cfg_filename.is_file():
+ LOG.fatal("Cannot find config file '%s'.", cfg_filename)
+ raise UsageError("Config file not found.")
- if not cfg_filename.is_file():
- LOG.fatal("Cannot find config file '%s'.", cfg_filename)
- raise UsageError("Config file not found.")
+ return cfg_filename
- return cfg_filename
+ filename = cfg_filename
search_paths = [self.project_dir, self.config_dir]
for path in search_paths:
- if (path / filename).is_file():
+ if path is not None and (path / filename).is_file():
return path / filename
LOG.fatal("Configuration file '%s' not found.\nDirectories searched: %s",
if Path(fname).is_absolute():
configfile = Path(fname)
else:
- configfile = self._find_config_file(loader.construct_scalar(node))
+ configfile = self.find_config_file(loader.construct_scalar(node))
if configfile.suffix != '.yaml':
- LOG.format("Format error while reading '%s': only YAML format supported.",
- configfile)
+ LOG.fatal("Format error while reading '%s': only YAML format supported.",
+ configfile)
raise UsageError("Cannot handle config file format.")
return yaml.safe_load(configfile.read_text(encoding='utf-8'))
projects / nominatim.git / blobdiff