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 import nominatim.api as napi
  15 from nominatim.api.v1.format import dispatch as formatting
  16
  17 CONTENT_TYPE = {
  18   'text': 'text/plain; charset=utf-8',
  19   'xml': 'text/xml; charset=utf-8',
  20   'jsonp': 'application/javascript'
  21 }
  22
  23
  24 class ASGIAdaptor(abc.ABC):
  25     """ Adapter class for the different ASGI frameworks.
  26         Wraps functionality over concrete requests and responses.
  27     """
  28
  29     @abc.abstractmethod
  30     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  31         """ Return an input parameter as a string. If the parameter was
  32             not provided, return the 'default' value.
  33         """
  34
  35     @abc.abstractmethod
  36     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  37         """ Return a HTTP header parameter as a string. If the parameter was
  38             not provided, return the 'default' value.
  39         """
  40
  41
  42     @abc.abstractmethod
  43     def error(self, msg: str) -> Exception:
  44         """ Construct an appropriate exception from the given error message.
  45             The exception must result in a HTTP 400 error.
  46         """
  47
  48
  49     @abc.abstractmethod
  50     def create_response(self, status: int, output: str, content_type: str) -> Any:
  51         """ Create a response from the given parameters. The result will
  52             be returned by the endpoint functions. The adaptor may also
  53             return None when the response is created internally with some
  54             different means.
  55
  56             The response must return the HTTP given status code 'status', set
  57             the HTTP content-type headers to the string provided and the
  58             body of the response to 'output'.
  59         """
  60
  61
  62     def build_response(self, output: str, media_type: str, status: int = 200) -> Any:
  63         """ Create a response from the given output. Wraps a JSONP function
  64             around the response, if necessary.
  65         """
  66         if media_type == 'json' and status == 200:
  67             jsonp = self.get('json_callback')
  68             if jsonp is not None:
  69                 if any(not part.isidentifier() for part in jsonp.split('.')):
  70                     raise self.error('Invalid json_callback value')
  71                 output = f"{jsonp}({output})"
  72                 media_type = 'jsonp'
  73
  74         return self.create_response(status, output,
  75                                     CONTENT_TYPE.get(media_type, 'application/json'))
  76
  77
  78     def get_int(self, name: str, default: Optional[int] = None) -> int:
  79         """ Return an input parameter as an int. Raises an exception if
  80             the parameter is given but not in an integer format.
  81
  82             If 'default' is given, then it will be returned when the parameter
  83             is missing completely. When 'default' is None, an error will be
  84             raised on a missing parameter.
  85         """
  86         value = self.get(name)
  87
  88         if value is None:
  89             if default is not None:
  90                 return default
  91
  92             raise self.error(f"Parameter '{name}' missing.")
  93
  94         try:
  95             return int(value)
  96         except ValueError as exc:
  97             raise self.error(f"Parameter '{name}' must be a number.") from exc
  98
  99
 100     def get_bool(self, name: str, default: Optional[bool] = None) -> bool:
 101         """ Return an input parameter as bool. Only '0' is accepted as
 102             an input for 'false' all other inputs will be interpreted as 'true'.
 103
 104             If 'default' is given, then it will be returned when the parameter
 105             is missing completely. When 'default' is None, an error will be
 106             raised on a missing parameter.
 107         """
 108         value = self.get(name)
 109
 110         if value is None:
 111             if default is not None:
 112                 return default
 113
 114             raise self.error(f"Parameter '{name}' missing.")
 115
 116         return value != '0'
 117
 118
 119 def parse_format(params: ASGIAdaptor, result_type: Type[Any], default: str) -> str:
 120     """ Get and check the 'format' parameter and prepare the formatter.
 121         `fmtter` is a formatter and `default` the
 122         format value to assume when no parameter is present.
 123     """
 124     fmt = params.get('format', default=default)
 125     assert fmt is not None
 126
 127     if not formatting.supports_format(result_type, fmt):
 128         raise params.error("Parameter 'format' must be one of: " +
 129                            ', '.join(formatting.list_formats(result_type)))
 130
 131     return fmt
 132
 133
 134 async def status_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
 135     """ Server glue for /status endpoint. See API docs for details.
 136     """
 137     result = await api.status()
 138
 139     fmt = parse_format(params, napi.StatusResult, 'text')
 140
 141     if fmt == 'text' and result.status:
 142         status_code = 500
 143     else:
 144         status_code = 200
 145
 146     return params.build_response(formatting.format_result(result, fmt), fmt,
 147                                  status=status_code)
 148
 149 EndpointFunc = Callable[[napi.NominatimAPIAsync, ASGIAdaptor], Any]
 150
 151 ROUTES = [
 152     ('status', status_endpoint)
 153 ]