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