nominatim/config.py

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