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