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