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