nominatim/config.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Nominatim configuration accessor.
   9 """
  10 from typing import Dict, Any, List, Mapping, Optional, Union
  11 import logging
  12 import os
  13 from pathlib import Path
  14 import json
  15 import yaml
  16
  17 from dotenv import dotenv_values
  18
  19 from nominatim.errors import UsageError
  20
  21 PathOrStr = Union[str, os.PathLike[str]]
  22
  23 LOG = logging.getLogger()
  24 CONFIG_CACHE : Dict[str, Any] = {}
  25
  26 def flatten_config_list(content: Any, section: str = '') -> List[Any]:
  27     """ Flatten YAML configuration lists that contain include sections
  28         which are lists themselves.
  29     """
  30     if not content:
  31         return []
  32
  33     if not isinstance(content, list):
  34         raise UsageError(f"List expected in section '{section}'.")
  35
  36     output = []
  37     for ele in content:
  38         if isinstance(ele, list):
  39             output.extend(flatten_config_list(ele, section))
  40         else:
  41             output.append(ele)
  42
  43     return output
  44
  45
  46 class Configuration:
  47     """ Load and manage the project configuration.
  48
  49         Nominatim uses dotenv to configure the software. Configuration options
  50         are resolved in the following order:
  51
  52          * from the OS environment (or the dirctionary given in `environ`
  53          * from the .env file in the project directory of the installation
  54          * from the default installation in the configuration directory
  55
  56         All Nominatim configuration options are prefixed with 'NOMINATIM_' to
  57         avoid conflicts with other environment variables.
  58     """
  59
  60     def __init__(self, project_dir: Path, config_dir: Path,
  61                  environ: Optional[Mapping[str, str]] = None) -> None:
  62         self.environ = environ or os.environ
  63         self.project_dir = project_dir
  64         self.config_dir = config_dir
  65         self._config = dotenv_values(str((config_dir / 'env.defaults').resolve()))
  66         if project_dir is not None and (project_dir / '.env').is_file():
  67             self._config.update(dotenv_values(str((project_dir / '.env').resolve())))
  68
  69         class _LibDirs:
  70             pass
  71
  72         self.lib_dir = _LibDirs()
  73
  74
  75     def set_libdirs(self, **kwargs: PathOrStr) -> None:
  76         """ Set paths to library functions and data.
  77         """
  78         for key, value in kwargs.items():
  79             setattr(self.lib_dir, key, Path(value).resolve())
  80
  81
  82     def __getattr__(self, name: str) -> str:
  83         name = 'NOMINATIM_' + name
  84
  85         if name in self.environ:
  86             return self.environ[name]
  87
  88         return self._config[name] or ''
  89
  90
  91     def get_bool(self, name: str) -> bool:
  92         """ Return the given configuration parameter as a boolean.
  93             Values of '1', 'yes' and 'true' are accepted as truthy values,
  94             everything else is interpreted as false.
  95         """
  96         return getattr(self, name).lower() in ('1', 'yes', 'true')
  97
  98
  99     def get_int(self, name: str) -> int:
 100         """ Return the given configuration parameter as an int.
 101         """
 102         try:
 103             return int(getattr(self, name))
 104         except ValueError as exp:
 105             LOG.fatal("Invalid setting NOMINATIM_%s. Needs to be a number.", name)
 106             raise UsageError("Configuration error.") from exp
 107
 108
 109     def get_str_list(self, name: str) -> Optional[List[str]]:
 110         """ Return the given configuration parameter as a list of strings.
 111             The values are assumed to be given as a comma-sparated list and
 112             will be stripped before returning them. On empty values None
 113             is returned.
 114         """
 115         raw = getattr(self, name)
 116
 117         return [v.strip() for v in raw.split(',')] if raw else None
 118
 119
 120     def get_path(self, name: str) -> Optional[Path]:
 121         """ Return the given configuration parameter as a Path.
 122             If a relative path is configured, then the function converts this
 123             into an absolute path with the project directory as root path.
 124             If the configuration is unset, None is returned.
 125         """
 126         value = getattr(self, name)
 127         if not value:
 128             return None
 129
 130         cfgpath = Path(value)
 131
 132         if not cfgpath.is_absolute():
 133             cfgpath = self.project_dir / cfgpath
 134
 135         return cfgpath.resolve()
 136
 137
 138     def get_libpq_dsn(self) -> str:
 139         """ Get configured database DSN converted into the key/value format
 140             understood by libpq and psycopg.
 141         """
 142         dsn = self.DATABASE_DSN
 143
 144         def quote_param(param: str) -> str:
 145             key, val = param.split('=')
 146             val = val.replace('\\', '\\\\').replace("'", "\\'")
 147             if ' ' in val:
 148                 val = "'" + val + "'"
 149             return key + '=' + val
 150
 151         if dsn.startswith('pgsql:'):
 152             # Old PHP DSN format. Convert before returning.
 153             return ' '.join([quote_param(p) for p in dsn[6:].split(';')])
 154
 155         return dsn
 156
 157
 158     def get_import_style_file(self) -> Path:
 159         """ Return the import style file as a path object. Translates the
 160             name of the standard styles automatically into a file in the
 161             config style.
 162         """
 163         style = getattr(self, 'IMPORT_STYLE')
 164
 165         if style in ('admin', 'street', 'address', 'full', 'extratags'):
 166             return self.config_dir / f'import-{style}.style'
 167
 168         return self.find_config_file('', 'IMPORT_STYLE')
 169
 170
 171     def get_os_env(self) -> Dict[str, Optional[str]]:
 172         """ Return a copy of the OS environment with the Nominatim configuration
 173             merged in.
 174         """
 175         env = dict(self._config)
 176         env.update(self.environ)
 177
 178         return env
 179
 180
 181     def load_sub_configuration(self, filename: PathOrStr,
 182                                config: Optional[str] = None) -> Any:
 183         """ Load additional configuration from a file. `filename` is the name
 184             of the configuration file. The file is first searched in the
 185             project directory and then in the global settings dirctory.
 186
 187             If `config` is set, then the name of the configuration file can
 188             be additionally given through a .env configuration option. When
 189             the option is set, then the file will be exclusively loaded as set:
 190             if the name is an absolute path, the file name is taken as is,
 191             if the name is relative, it is taken to be relative to the
 192             project directory.
 193
 194             The format of the file is determined from the filename suffix.
 195             Currently only files with extension '.yaml' are supported.
 196
 197             YAML files support a special '!include' construct. When the
 198             directive is given, the value is taken to be a filename, the file
 199             is loaded using this function and added at the position in the
 200             configuration tree.
 201         """
 202         configfile = self.find_config_file(filename, config)
 203
 204         if str(configfile) in CONFIG_CACHE:
 205             return CONFIG_CACHE[str(configfile)]
 206
 207         if configfile.suffix in ('.yaml', '.yml'):
 208             result = self._load_from_yaml(configfile)
 209         elif configfile.suffix == '.json':
 210             with configfile.open('r', encoding='utf-8') as cfg:
 211                 result = json.load(cfg)
 212         else:
 213             raise UsageError(f"Config file '{configfile}' has unknown format.")
 214
 215         CONFIG_CACHE[str(configfile)] = result
 216         return result
 217
 218
 219     def find_config_file(self, filename: PathOrStr,
 220                          config: Optional[str] = None) -> Path:
 221         """ Resolve the location of a configuration file given a filename and
 222             an optional configuration option with the file name.
 223             Raises a UsageError when the file cannot be found or is not
 224             a regular file.
 225         """
 226         if config is not None:
 227             cfg_value = getattr(self, config)
 228             if cfg_value:
 229                 cfg_filename = Path(cfg_value)
 230
 231                 if cfg_filename.is_absolute():
 232                     cfg_filename = cfg_filename.resolve()
 233
 234                     if not cfg_filename.is_file():
 235                         LOG.fatal("Cannot find config file '%s'.", cfg_filename)
 236                         raise UsageError("Config file not found.")
 237
 238                     return cfg_filename
 239
 240                 filename = cfg_filename
 241
 242
 243         search_paths = [self.project_dir, self.config_dir]
 244         for path in search_paths:
 245             if path is not None and (path / filename).is_file():
 246                 return path / filename
 247
 248         LOG.fatal("Configuration file '%s' not found.\nDirectories searched: %s",
 249                   filename, search_paths)
 250         raise UsageError("Config file not found.")
 251
 252
 253     def _load_from_yaml(self, cfgfile: Path) -> Any:
 254         """ Load a YAML configuration file. This installs a special handler that
 255             allows to include other YAML files using the '!include' operator.
 256         """
 257         yaml.add_constructor('!include', self._yaml_include_representer,
 258                              Loader=yaml.SafeLoader)
 259         return yaml.safe_load(cfgfile.read_text(encoding='utf-8'))
 260
 261
 262     def _yaml_include_representer(self, loader: Any, node: yaml.Node) -> Any:
 263         """ Handler for the '!include' operator in YAML files.
 264
 265             When the filename is relative, then the file is first searched in the
 266             project directory and then in the global settings dirctory.
 267         """
 268         fname = loader.construct_scalar(node)
 269
 270         if Path(fname).is_absolute():
 271             configfile = Path(fname)
 272         else:
 273             configfile = self.find_config_file(loader.construct_scalar(node))
 274
 275         if configfile.suffix != '.yaml':
 276             LOG.fatal("Format error while reading '%s': only YAML format supported.",
 277                       configfile)
 278             raise UsageError("Cannot handle config file format.")
 279
 280         return yaml.safe_load(configfile.read_text(encoding='utf-8'))