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