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