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