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