src/nominatim_core/config.py

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