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