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 logging
  12 import os
  13 from pathlib import Path
  14 import json
  15 import yaml
  16
  17 from dotenv import dotenv_values
  18
  19 from nominatim.typing import StrPath
  20 from nominatim.errors import UsageError
  21
  22 LOG = logging.getLogger()
  23 CONFIG_CACHE : Dict[str, Any] = {}
  24
  25 def flatten_config_list(content: Any, section: str = '') -> List[Any]:
  26     """ Flatten YAML configuration lists that contain include sections
  27         which are lists themselves.
  28     """
  29     if not content:
  30         return []
  31
  32     if not isinstance(content, list):
  33         raise UsageError(f"List expected in section '{section}'.")
  34
  35     output = []
  36     for ele in content:
  37         if isinstance(ele, list):
  38             output.extend(flatten_config_list(ele, section))
  39         else:
  40             output.append(ele)
  41
  42     return output
  43
  44
  45 class Configuration:
  46     """ Load and manage the project configuration.
  47
  48         Nominatim uses dotenv to configure the software. Configuration options
  49         are resolved in the following order:
  50
  51          * from the OS environment (or the dirctionary given in `environ`
  52          * from the .env file in the project directory of the installation
  53          * from the default installation in the configuration directory
  54
  55         All Nominatim configuration options are prefixed with 'NOMINATIM_' to
  56         avoid conflicts with other environment variables.
  57     """
  58
  59     def __init__(self, project_dir: Path, config_dir: Path,
  60                  environ: Optional[Mapping[str, str]] = None) -> None:
  61         self.environ = environ or os.environ
  62         self.project_dir = project_dir
  63         self.config_dir = config_dir
  64         self._config = dotenv_values(str((config_dir / 'env.defaults').resolve()))
  65         if project_dir is not None and (project_dir / '.env').is_file():
  66             self._config.update(dotenv_values(str((project_dir / '.env').resolve())))
  67
  68         class _LibDirs:
  69             module: Path
  70             osm2pgsql: Path
  71             php: Path
  72             sql: Path
  73             data: Path
  74
  75         self.lib_dir = _LibDirs()
  76
  77
  78     def set_libdirs(self, **kwargs: StrPath) -> None:
  79         """ Set paths to library functions and data.
  80         """
  81         for key, value in kwargs.items():
  82             setattr(self.lib_dir, key, Path(value).resolve())
  83
  84
  85     def __getattr__(self, name: str) -> str:
  86         name = 'NOMINATIM_' + name
  87
  88         if name in self.environ:
  89             return self.environ[name]
  90
  91         return self._config[name] or ''
  92
  93
  94     def get_bool(self, name: str) -> bool:
  95         """ Return the given configuration parameter as a boolean.
  96             Values of '1', 'yes' and 'true' are accepted as truthy values,
  97             everything else is interpreted as false.
  98         """
  99         return getattr(self, name).lower() in ('1', 'yes', 'true')
 100
 101
 102     def get_int(self, name: str) -> int:
 103         """ Return the given configuration parameter as an int.
 104         """
 105         try:
 106             return int(getattr(self, name))
 107         except ValueError as exp:
 108             LOG.fatal("Invalid setting NOMINATIM_%s. Needs to be a number.", name)
 109             raise UsageError("Configuration error.") from exp
 110
 111
 112     def get_str_list(self, name: str) -> Optional[List[str]]:
 113         """ Return the given configuration parameter as a list of strings.
 114             The values are assumed to be given as a comma-sparated list and
 115             will be stripped before returning them. On empty values None
 116             is returned.
 117         """
 118         raw = getattr(self, name)
 119
 120         return [v.strip() for v in raw.split(',')] if raw else None
 121
 122
 123     def get_path(self, name: str) -> Optional[Path]:
 124         """ Return the given configuration parameter as a Path.
 125             If a relative path is configured, then the function converts this
 126             into an absolute path with the project directory as root path.
 127             If the configuration is unset, None is returned.
 128         """
 129         value = getattr(self, name)
 130         if not value:
 131             return None
 132
 133         cfgpath = Path(value)
 134
 135         if not cfgpath.is_absolute():
 136             cfgpath = self.project_dir / cfgpath
 137
 138         return cfgpath.resolve()
 139
 140
 141     def get_libpq_dsn(self) -> str:
 142         """ Get configured database DSN converted into the key/value format
 143             understood by libpq and psycopg.
 144         """
 145         dsn = self.DATABASE_DSN
 146
 147         def quote_param(param: str) -> str:
 148             key, val = param.split('=')
 149             val = val.replace('\\', '\\\\').replace("'", "\\'")
 150             if ' ' in val:
 151                 val = "'" + val + "'"
 152             return key + '=' + val
 153
 154         if dsn.startswith('pgsql:'):
 155             # Old PHP DSN format. Convert before returning.
 156             return ' '.join([quote_param(p) for p in dsn[6:].split(';')])
 157
 158         return dsn
 159
 160
 161     def get_import_style_file(self) -> Path:
 162         """ Return the import style file as a path object. Translates the
 163             name of the standard styles automatically into a file in the
 164             config style.
 165         """
 166         style = getattr(self, 'IMPORT_STYLE')
 167
 168         if style in ('admin', 'street', 'address', 'full', 'extratags'):
 169             return self.config_dir / f'import-{style}.style'
 170
 171         return self.find_config_file('', 'IMPORT_STYLE')
 172
 173
 174     def get_os_env(self) -> Dict[str, Optional[str]]:
 175         """ Return a copy of the OS environment with the Nominatim configuration
 176             merged in.
 177         """
 178         env = dict(self._config)
 179         env.update(self.environ)
 180
 181         return env
 182
 183
 184     def load_sub_configuration(self, filename: StrPath,
 185                                config: Optional[str] = None) -> Any:
 186         """ Load additional configuration from a file. `filename` is the name
 187             of the configuration file. The file is first searched in the
 188             project directory and then in the global settings directory.
 189
 190             If `config` is set, then the name of the configuration file can
 191             be additionally given through a .env configuration option. When
 192             the option is set, then the file will be exclusively loaded as set:
 193             if the name is an absolute path, the file name is taken as is,
 194             if the name is relative, it is taken to be relative to the
 195             project directory.
 196
 197             The format of the file is determined from the filename suffix.
 198             Currently only files with extension '.yaml' are supported.
 199
 200             YAML files support a special '!include' construct. When the
 201             directive is given, the value is taken to be a filename, the file
 202             is loaded using this function and added at the position in the
 203             configuration tree.
 204         """
 205         configfile = self.find_config_file(filename, config)
 206
 207         if str(configfile) in CONFIG_CACHE:
 208             return CONFIG_CACHE[str(configfile)]
 209
 210         if configfile.suffix in ('.yaml', '.yml'):
 211             result = self._load_from_yaml(configfile)
 212         elif configfile.suffix == '.json':
 213             with configfile.open('r', encoding='utf-8') as cfg:
 214                 result = json.load(cfg)
 215         else:
 216             raise UsageError(f"Config file '{configfile}' has unknown format.")
 217
 218         CONFIG_CACHE[str(configfile)] = result
 219         return result
 220
 221
 222     def find_config_file(self, filename: StrPath,
 223                          config: Optional[str] = None) -> Path:
 224         """ Resolve the location of a configuration file given a filename and
 225             an optional configuration option with the file name.
 226             Raises a UsageError when the file cannot be found or is not
 227             a regular file.
 228         """
 229         if config is not None:
 230             cfg_value = getattr(self, config)
 231             if cfg_value:
 232                 cfg_filename = Path(cfg_value)
 233
 234                 if cfg_filename.is_absolute():
 235                     cfg_filename = cfg_filename.resolve()
 236
 237                     if not cfg_filename.is_file():
 238                         LOG.fatal("Cannot find config file '%s'.", cfg_filename)
 239                         raise UsageError("Config file not found.")
 240
 241                     return cfg_filename
 242
 243                 filename = cfg_filename
 244
 245
 246         search_paths = [self.project_dir, self.config_dir]
 247         for path in search_paths:
 248             if path is not None and (path / filename).is_file():
 249                 return path / filename
 250
 251         LOG.fatal("Configuration file '%s' not found.\nDirectories searched: %s",
 252                   filename, search_paths)
 253         raise UsageError("Config file not found.")
 254
 255
 256     def _load_from_yaml(self, cfgfile: Path) -> Any:
 257         """ Load a YAML configuration file. This installs a special handler that
 258             allows to include other YAML files using the '!include' operator.
 259         """
 260         yaml.add_constructor('!include', self._yaml_include_representer,
 261                              Loader=yaml.SafeLoader)
 262         return yaml.safe_load(cfgfile.read_text(encoding='utf-8'))
 263
 264
 265     def _yaml_include_representer(self, loader: Any, node: yaml.Node) -> Any:
 266         """ Handler for the '!include' operator in YAML files.
 267
 268             When the filename is relative, then the file is first searched in the
 269             project directory and then in the global settings directory.
 270         """
 271         fname = loader.construct_scalar(node)
 272
 273         if Path(fname).is_absolute():
 274             configfile = Path(fname)
 275         else:
 276             configfile = self.find_config_file(loader.construct_scalar(node))
 277
 278         if configfile.suffix != '.yaml':
 279             LOG.fatal("Format error while reading '%s': only YAML format supported.",
 280                       configfile)
 281             raise UsageError("Cannot handle config file format.")
 282
 283         return yaml.safe_load(configfile.read_text(encoding='utf-8'))