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