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