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
  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   'jsonp': 'application/javascript',
  23   'debug': 'text/html; charset=utf-8'
  24 }
  25
  26
  27 class ASGIAdaptor(abc.ABC):
  28     """ Adapter class for the different ASGI frameworks.
  29         Wraps functionality over concrete requests and responses.
  30     """
  31
  32     @abc.abstractmethod
  33     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  34         """ Return an input parameter as a string. If the parameter was
  35             not provided, return the 'default' value.
  36         """
  37
  38     @abc.abstractmethod
  39     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  40         """ Return a HTTP header parameter as a string. If the parameter was
  41             not provided, return the 'default' value.
  42         """
  43
  44
  45     @abc.abstractmethod
  46     def error(self, msg: str, status: int = 400) -> Exception:
  47         """ Construct an appropriate exception from the given error message.
  48             The exception must result in a HTTP error with the given status.
  49         """
  50
  51
  52     @abc.abstractmethod
  53     def create_response(self, status: int, output: str, content_type: str) -> Any:
  54         """ Create a response from the given parameters. The result will
  55             be returned by the endpoint functions. The adaptor may also
  56             return None when the response is created internally with some
  57             different means.
  58
  59             The response must return the HTTP given status code 'status', set
  60             the HTTP content-type headers to the string provided and the
  61             body of the response to 'output'.
  62         """
  63
  64
  65     @abc.abstractmethod
  66     def config(self) -> Configuration:
  67         """ Return the current configuration object.
  68         """
  69
  70
  71     def build_response(self, output: str, media_type: str, status: int = 200) -> Any:
  72         """ Create a response from the given output. Wraps a JSONP function
  73             around the response, if necessary.
  74         """
  75         if media_type == 'json' and status == 200:
  76             jsonp = self.get('json_callback')
  77             if jsonp is not None:
  78                 if any(not part.isidentifier() for part in jsonp.split('.')):
  79                     raise self.error('Invalid json_callback value')
  80                 output = f"{jsonp}({output})"
  81                 media_type = 'jsonp'
  82
  83         return self.create_response(status, output,
  84                                     CONTENT_TYPE.get(media_type, 'application/json'))
  85
  86
  87     def get_int(self, name: str, default: Optional[int] = None) -> int:
  88         """ Return an input parameter as an int. Raises an exception if
  89             the parameter is given but not in an integer format.
  90
  91             If 'default' is given, then it will be returned when the parameter
  92             is missing completely. When 'default' is None, an error will be
  93             raised on a missing parameter.
  94         """
  95         value = self.get(name)
  96
  97         if value is None:
  98             if default is not None:
  99                 return default
 100
 101             raise self.error(f"Parameter '{name}' missing.")
 102
 103         try:
 104             return int(value)
 105         except ValueError as exc:
 106             raise self.error(f"Parameter '{name}' must be a number.") from exc
 107
 108
 109     def get_bool(self, name: str, default: Optional[bool] = None) -> bool:
 110         """ Return an input parameter as bool. Only '0' is accepted as
 111             an input for 'false' all other inputs will be interpreted as 'true'.
 112
 113             If 'default' is given, then it will be returned when the parameter
 114             is missing completely. When 'default' is None, an error will be
 115             raised on a missing parameter.
 116         """
 117         value = self.get(name)
 118
 119         if value is None:
 120             if default is not None:
 121                 return default
 122
 123             raise self.error(f"Parameter '{name}' missing.")
 124
 125         return value != '0'
 126
 127
 128     def get_accepted_languages(self) -> str:
 129         """ Return the accepted langauges.
 130         """
 131         return self.get('accept-language')\
 132                or self.get_header('http_accept_language')\
 133                or self.config().DEFAULT_LANGUAGE
 134
 135
 136     def setup_debugging(self) -> bool:
 137         """ Set up collection of debug information if requested.
 138
 139             Return True when debugging was requested.
 140         """
 141         if self.get_bool('debug', False):
 142             loglib.set_log_output('html')
 143             return True
 144
 145         return False
 146
 147
 148 def parse_format(params: ASGIAdaptor, result_type: Type[Any], default: str) -> str:
 149     """ Get and check the 'format' parameter and prepare the formatter.
 150         `fmtter` is a formatter and `default` the
 151         format value to assume when no parameter is present.
 152     """
 153     fmt = params.get('format', default=default)
 154     assert fmt is not None
 155
 156     if not formatting.supports_format(result_type, fmt):
 157         raise params.error("Parameter 'format' must be one of: " +
 158                            ', '.join(formatting.list_formats(result_type)))
 159
 160     return fmt
 161
 162
 163 async def status_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
 164     """ Server glue for /status endpoint. See API docs for details.
 165     """
 166     result = await api.status()
 167
 168     fmt = parse_format(params, napi.StatusResult, 'text')
 169
 170     if fmt == 'text' and result.status:
 171         status_code = 500
 172     else:
 173         status_code = 200
 174
 175     return params.build_response(formatting.format_result(result, fmt, {}), fmt,
 176                                  status=status_code)
 177
 178
 179 async def details_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
 180     """ Server glue for /details endpoint. See API docs for details.
 181     """
 182     place_id = params.get_int('place_id', 0)
 183     place: napi.PlaceRef
 184     if place_id:
 185         place = napi.PlaceID(place_id)
 186     else:
 187         osmtype = params.get('osmtype')
 188         if osmtype is None:
 189             raise params.error("Missing ID parameter 'place_id' or 'osmtype'.")
 190         place = napi.OsmID(osmtype, params.get_int('osmid'), params.get('class'))
 191
 192     debug = params.setup_debugging()
 193
 194     details = napi.LookupDetails(address_details=params.get_bool('addressdetails', False),
 195                                  linked_places=params.get_bool('linkedplaces', False),
 196                                  parented_places=params.get_bool('hierarchy', False),
 197                                  keywords=params.get_bool('keywords', False))
 198
 199     if params.get_bool('polygon_geojson', False):
 200         details.geometry_output = napi.GeometryFormat.GEOJSON
 201
 202     locales = napi.Locales.from_accept_languages(params.get_accepted_languages())
 203
 204     result = await api.lookup(place, details)
 205
 206     if debug:
 207         return params.build_response(loglib.get_and_disable(), 'debug')
 208
 209     if result is None:
 210         raise params.error('No place with that OSM ID found.', status=404)
 211
 212     output = formatting.format_result(
 213                  result,
 214                  'details-json',
 215                  {'locales': locales,
 216                   'group_hierarchy': params.get_bool('group_hierarchy', False),
 217                   'icon_base_url': params.config().MAPICON_URL})
 218
 219     return params.build_response(output, 'json')
 220
 221
 222 EndpointFunc = Callable[[napi.NominatimAPIAsync, ASGIAdaptor], Any]
 223
 224 ROUTES = [
 225     ('status', status_endpoint),
 226     ('details', details_endpoint)
 227 ]