nominatim/api/v1/server_glue.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2023 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Generic part of the server implementation of the v1 API.
   9 Combine with the scaffolding provided for the various Python ASGI frameworks.
  10 """
  11 from typing import Optional, Any, Type, Callable, NoReturn, Dict, cast
  12 from functools import reduce
  13 import abc
  14 import math
  15
  16 from nominatim.config import Configuration
  17 import nominatim.api as napi
  18 import nominatim.api.logging as loglib
  19 from nominatim.api.v1.format import dispatch as formatting
  20 from nominatim.api.v1 import helpers
  21
  22 CONTENT_TYPE = {
  23   'text': 'text/plain; charset=utf-8',
  24   'xml': 'text/xml; charset=utf-8',
  25   'debug': 'text/html; charset=utf-8'
  26 }
  27
  28 class ASGIAdaptor(abc.ABC):
  29     """ Adapter class for the different ASGI frameworks.
  30         Wraps functionality over concrete requests and responses.
  31     """
  32     content_type: str = 'text/plain; charset=utf-8'
  33
  34     @abc.abstractmethod
  35     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  36         """ Return an input parameter as a string. If the parameter was
  37             not provided, return the 'default' value.
  38         """
  39
  40     @abc.abstractmethod
  41     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  42         """ Return a HTTP header parameter as a string. If the parameter was
  43             not provided, return the 'default' value.
  44         """
  45
  46
  47     @abc.abstractmethod
  48     def error(self, msg: str, status: int = 400) -> Exception:
  49         """ Construct an appropriate exception from the given error message.
  50             The exception must result in a HTTP error with the given status.
  51         """
  52
  53
  54     @abc.abstractmethod
  55     def create_response(self, status: int, output: str) -> Any:
  56         """ Create a response from the given parameters. The result will
  57             be returned by the endpoint functions. The adaptor may also
  58             return None when the response is created internally with some
  59             different means.
  60
  61             The response must return the HTTP given status code 'status', set
  62             the HTTP content-type headers to the string provided and the
  63             body of the response to 'output'.
  64         """
  65
  66
  67     @abc.abstractmethod
  68     def config(self) -> Configuration:
  69         """ Return the current configuration object.
  70         """
  71
  72
  73     def build_response(self, output: str, status: int = 200) -> Any:
  74         """ Create a response from the given output. Wraps a JSONP function
  75             around the response, if necessary.
  76         """
  77         if self.content_type == 'application/json' and status == 200:
  78             jsonp = self.get('json_callback')
  79             if jsonp is not None:
  80                 if any(not part.isidentifier() for part in jsonp.split('.')):
  81                     self.raise_error('Invalid json_callback value')
  82                 output = f"{jsonp}({output})"
  83                 self.content_type = 'application/javascript'
  84
  85         return self.create_response(status, output)
  86
  87
  88     def raise_error(self, msg: str, status: int = 400) -> NoReturn:
  89         """ Raise an exception resulting in the given HTTP status and
  90             message. The message will be formatted according to the
  91             output format chosen by the request.
  92         """
  93         if self.content_type == 'text/xml; charset=utf-8':
  94             msg = f"""<?xml version="1.0" encoding="UTF-8" ?>
  95                       <error>
  96                         <code>{status}</code>
  97                         <message>{msg}</message>
  98                       </error>
  99                    """
 100         elif self.content_type == 'application/json':
 101             msg = f"""{{"error":{{"code":{status},"message":"{msg}"}}}}"""
 102         elif self.content_type == 'text/html; charset=utf-8':
 103             loglib.log().section('Execution error')
 104             loglib.log().var_dump('Status', status)
 105             loglib.log().var_dump('Message', msg)
 106             msg = loglib.get_and_disable()
 107
 108         raise self.error(msg, status)
 109
 110
 111     def get_int(self, name: str, default: Optional[int] = None) -> int:
 112         """ Return an input parameter as an int. Raises an exception if
 113             the parameter is given but not in an integer format.
 114
 115             If 'default' is given, then it will be returned when the parameter
 116             is missing completely. When 'default' is None, an error will be
 117             raised on a missing parameter.
 118         """
 119         value = self.get(name)
 120
 121         if value is None:
 122             if default is not None:
 123                 return default
 124
 125             self.raise_error(f"Parameter '{name}' missing.")
 126
 127         try:
 128             intval = int(value)
 129         except ValueError:
 130             self.raise_error(f"Parameter '{name}' must be a number.")
 131
 132         return intval
 133
 134
 135     def get_float(self, name: str, default: Optional[float] = None) -> float:
 136         """ Return an input parameter as a flaoting-point number. Raises an
 137             exception if the parameter is given but not in an float format.
 138
 139             If 'default' is given, then it will be returned when the parameter
 140             is missing completely. When 'default' is None, an error will be
 141             raised on a missing parameter.
 142         """
 143         value = self.get(name)
 144
 145         if value is None:
 146             if default is not None:
 147                 return default
 148
 149             self.raise_error(f"Parameter '{name}' missing.")
 150
 151         try:
 152             fval = float(value)
 153         except ValueError:
 154             self.raise_error(f"Parameter '{name}' must be a number.")
 155
 156         if math.isnan(fval) or math.isinf(fval):
 157             self.raise_error(f"Parameter '{name}' must be a number.")
 158
 159         return fval
 160
 161
 162     def get_bool(self, name: str, default: Optional[bool] = None) -> bool:
 163         """ Return an input parameter as bool. Only '0' is accepted as
 164             an input for 'false' all other inputs will be interpreted as 'true'.
 165
 166             If 'default' is given, then it will be returned when the parameter
 167             is missing completely. When 'default' is None, an error will be
 168             raised on a missing parameter.
 169         """
 170         value = self.get(name)
 171
 172         if value is None:
 173             if default is not None:
 174                 return default
 175
 176             self.raise_error(f"Parameter '{name}' missing.")
 177
 178         return value != '0'
 179
 180
 181     def get_accepted_languages(self) -> str:
 182         """ Return the accepted languages.
 183         """
 184         return self.get('accept-language')\
 185                or self.get_header('http_accept_language')\
 186                or self.config().DEFAULT_LANGUAGE
 187
 188
 189     def setup_debugging(self) -> bool:
 190         """ Set up collection of debug information if requested.
 191
 192             Return True when debugging was requested.
 193         """
 194         if self.get_bool('debug', False):
 195             loglib.set_log_output('html')
 196             self.content_type = 'text/html; charset=utf-8'
 197             return True
 198
 199         return False
 200
 201
 202     def get_layers(self) -> Optional[napi.DataLayer]:
 203         """ Return a parsed version of the layer parameter.
 204         """
 205         param = self.get('layer', None)
 206         if param is None:
 207             return None
 208
 209         return cast(napi.DataLayer,
 210                     reduce(napi.DataLayer.__or__,
 211                            (getattr(napi.DataLayer, s.upper()) for s in param.split(','))))
 212
 213
 214     def parse_format(self, result_type: Type[Any], default: str) -> str:
 215         """ Get and check the 'format' parameter and prepare the formatter.
 216             `result_type` is the type of result to be returned by the function
 217             and `default` the format value to assume when no parameter is present.
 218         """
 219         fmt = self.get('format', default=default)
 220         assert fmt is not None
 221
 222         if not formatting.supports_format(result_type, fmt):
 223             self.raise_error("Parameter 'format' must be one of: " +
 224                               ', '.join(formatting.list_formats(result_type)))
 225
 226         self.content_type = CONTENT_TYPE.get(fmt, 'application/json')
 227         return fmt
 228
 229
 230     def parse_geometry_details(self, fmt: str) -> Dict[str, Any]:
 231         """ Create details strucutre from the supplied geometry parameters.
 232         """
 233         numgeoms = 0
 234         output = napi.GeometryFormat.NONE
 235         if self.get_bool('polygon_geojson', False):
 236             output |= napi.GeometryFormat.GEOJSON
 237             numgeoms += 1
 238         if fmt not in ('geojson', 'geocodejson'):
 239             if self.get_bool('polygon_text', False):
 240                 output |= napi.GeometryFormat.TEXT
 241                 numgeoms += 1
 242             if self.get_bool('polygon_kml', False):
 243                 output |= napi.GeometryFormat.KML
 244                 numgeoms += 1
 245             if self.get_bool('polygon_svg', False):
 246                 output |= napi.GeometryFormat.SVG
 247                 numgeoms += 1
 248
 249         if numgeoms > self.config().get_int('POLYGON_OUTPUT_MAX_TYPES'):
 250             self.raise_error('Too many polgyon output options selected.')
 251
 252         return {'address_details': True,
 253                 'geometry_simplification': self.get_float('polygon_threshold', 0.0),
 254                 'geometry_output': output
 255                }
 256
 257
 258 async def status_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
 259     """ Server glue for /status endpoint. See API docs for details.
 260     """
 261     result = await api.status()
 262
 263     fmt = params.parse_format(napi.StatusResult, 'text')
 264
 265     if fmt == 'text' and result.status:
 266         status_code = 500
 267     else:
 268         status_code = 200
 269
 270     return params.build_response(formatting.format_result(result, fmt, {}),
 271                                  status=status_code)
 272
 273
 274 async def details_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
 275     """ Server glue for /details endpoint. See API docs for details.
 276     """
 277     fmt = params.parse_format(napi.DetailedResult, 'json')
 278     place_id = params.get_int('place_id', 0)
 279     place: napi.PlaceRef
 280     if place_id:
 281         place = napi.PlaceID(place_id)
 282     else:
 283         osmtype = params.get('osmtype')
 284         if osmtype is None:
 285             params.raise_error("Missing ID parameter 'place_id' or 'osmtype'.")
 286         place = napi.OsmID(osmtype, params.get_int('osmid'), params.get('class'))
 287
 288     debug = params.setup_debugging()
 289
 290     locales = napi.Locales.from_accept_languages(params.get_accepted_languages())
 291
 292     result = await api.details(place,
 293                                address_details=params.get_bool('addressdetails', False),
 294                                linked_places=params.get_bool('linkedplaces', False),
 295                                parented_places=params.get_bool('hierarchy', False),
 296                                keywords=params.get_bool('keywords', False),
 297                                geometry_output = napi.GeometryFormat.GEOJSON
 298                                                  if params.get_bool('polygon_geojson', False)
 299                                                  else napi.GeometryFormat.NONE
 300                               )
 301
 302     if debug:
 303         return params.build_response(loglib.get_and_disable())
 304
 305     if result is None:
 306         params.raise_error('No place with that OSM ID found.', status=404)
 307
 308     output = formatting.format_result(result, fmt,
 309                  {'locales': locales,
 310                   'group_hierarchy': params.get_bool('group_hierarchy', False),
 311                   'icon_base_url': params.config().MAPICON_URL})
 312
 313     return params.build_response(output)
 314
 315
 316 async def reverse_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
 317     """ Server glue for /reverse endpoint. See API docs for details.
 318     """
 319     fmt = params.parse_format(napi.ReverseResults, 'xml')
 320     debug = params.setup_debugging()
 321     coord = napi.Point(params.get_float('lon'), params.get_float('lat'))
 322     locales = napi.Locales.from_accept_languages(params.get_accepted_languages())
 323
 324     details = params.parse_geometry_details(fmt)
 325     details['max_rank'] = helpers.zoom_to_rank(params.get_int('zoom', 18))
 326     details['layers'] = params.get_layers()
 327
 328     result = await api.reverse(coord, **details)
 329
 330     if debug:
 331         return params.build_response(loglib.get_and_disable())
 332
 333     fmt_options = {'locales': locales,
 334                    'extratags': params.get_bool('extratags', False),
 335                    'namedetails': params.get_bool('namedetails', False),
 336                    'addressdetails': params.get_bool('addressdetails', True)}
 337
 338     output = formatting.format_result(napi.ReverseResults([result] if result else []),
 339                                       fmt, fmt_options)
 340
 341     return params.build_response(output)
 342
 343
 344 async def lookup_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
 345     """ Server glue for /lookup endpoint. See API docs for details.
 346     """
 347     fmt = params.parse_format(napi.SearchResults, 'xml')
 348     debug = params.setup_debugging()
 349     locales = napi.Locales.from_accept_languages(params.get_accepted_languages())
 350     details = params.parse_geometry_details(fmt)
 351
 352     places = []
 353     for oid in (params.get('osm_ids') or '').split(','):
 354         oid = oid.strip()
 355         if len(oid) > 1 and oid[0] in 'RNWrnw' and oid[1:].isdigit():
 356             places.append(napi.OsmID(oid[0], int(oid[1:])))
 357
 358     if places:
 359         results = await api.lookup(places, **details)
 360     else:
 361         results = napi.SearchResults()
 362
 363     if debug:
 364         return params.build_response(loglib.get_and_disable())
 365
 366     fmt_options = {'locales': locales,
 367                    'extratags': params.get_bool('extratags', False),
 368                    'namedetails': params.get_bool('namedetails', False),
 369                    'addressdetails': params.get_bool('addressdetails', True)}
 370
 371     output = formatting.format_result(results, fmt, fmt_options)
 372
 373     return params.build_response(output)
 374
 375 EndpointFunc = Callable[[napi.NominatimAPIAsync, ASGIAdaptor], Any]
 376
 377 ROUTES = [
 378     ('status', status_endpoint),
 379     ('details', details_endpoint),
 380     ('reverse', reverse_endpoint),
 381     ('lookup', lookup_endpoint)
 382 ]