src/nominatim_api/server/asgi_adaptor.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2024 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Base abstraction for implementing based on different ASGI frameworks.
   9 """
  10 from typing import Optional, Any, NoReturn, Callable
  11 import abc
  12 import math
  13
  14 from ..config import Configuration
  15 from .. import logging as loglib
  16 from ..core import NominatimAPIAsync
  17 from ..result_formatting import FormatDispatcher
  18
  19 CONTENT_TEXT = 'text/plain; charset=utf-8'
  20 CONTENT_XML = 'text/xml; charset=utf-8'
  21 CONTENT_HTML = 'text/html; charset=utf-8'
  22 CONTENT_JSON = 'application/json; charset=utf-8'
  23
  24 CONTENT_TYPE = {'text': CONTENT_TEXT, 'xml': CONTENT_XML, 'debug': CONTENT_HTML}
  25
  26 class ASGIAdaptor(abc.ABC):
  27     """ Adapter class for the different ASGI frameworks.
  28         Wraps functionality over concrete requests and responses.
  29     """
  30     content_type: str = CONTENT_TEXT
  31
  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, num_results: int) -> 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 base_uri(self) -> str:
  68         """ Return the URI of the original request.
  69         """
  70
  71
  72     @abc.abstractmethod
  73     def config(self) -> Configuration:
  74         """ Return the current configuration object.
  75         """
  76
  77
  78     @abc.abstractmethod
  79     def formatting(self) -> FormatDispatcher:
  80         """ Return the formatting object to use.
  81         """
  82
  83
  84     def get_int(self, name: str, default: Optional[int] = None) -> int:
  85         """ Return an input parameter as an int. Raises an exception if
  86             the parameter is given but not in an integer format.
  87
  88             If 'default' is given, then it will be returned when the parameter
  89             is missing completely. When 'default' is None, an error will be
  90             raised on a missing parameter.
  91         """
  92         value = self.get(name)
  93
  94         if value is None:
  95             if default is not None:
  96                 return default
  97
  98             self.raise_error(f"Parameter '{name}' missing.")
  99
 100         try:
 101             intval = int(value)
 102         except ValueError:
 103             self.raise_error(f"Parameter '{name}' must be a number.")
 104
 105         return intval
 106
 107
 108     def get_float(self, name: str, default: Optional[float] = None) -> float:
 109         """ Return an input parameter as a flaoting-point number. Raises an
 110             exception if the parameter is given but not in an float format.
 111
 112             If 'default' is given, then it will be returned when the parameter
 113             is missing completely. When 'default' is None, an error will be
 114             raised on a missing parameter.
 115         """
 116         value = self.get(name)
 117
 118         if value is None:
 119             if default is not None:
 120                 return default
 121
 122             self.raise_error(f"Parameter '{name}' missing.")
 123
 124         try:
 125             fval = float(value)
 126         except ValueError:
 127             self.raise_error(f"Parameter '{name}' must be a number.")
 128
 129         if math.isnan(fval) or math.isinf(fval):
 130             self.raise_error(f"Parameter '{name}' must be a number.")
 131
 132         return fval
 133
 134
 135     def get_bool(self, name: str, default: Optional[bool] = None) -> bool:
 136         """ Return an input parameter as bool. Only '0' is accepted as
 137             an input for 'false' all other inputs will be interpreted as 'true'.
 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         return value != '0'
 152
 153
 154     def raise_error(self, msg: str, status: int = 400) -> NoReturn:
 155         """ Raise an exception resulting in the given HTTP status and
 156             message. The message will be formatted according to the
 157             output format chosen by the request.
 158         """
 159         if self.content_type == CONTENT_XML:
 160             msg = f"""<?xml version="1.0" encoding="UTF-8" ?>
 161                       <error>
 162                         <code>{status}</code>
 163                         <message>{msg}</message>
 164                       </error>
 165                    """
 166         elif self.content_type == CONTENT_JSON:
 167             msg = f"""{{"error":{{"code":{status},"message":"{msg}"}}}}"""
 168         elif self.content_type == CONTENT_HTML:
 169             loglib.log().section('Execution error')
 170             loglib.log().var_dump('Status', status)
 171             loglib.log().var_dump('Message', msg)
 172             msg = loglib.get_and_disable()
 173
 174         raise self.error(msg, status)
 175
 176
 177 EndpointFunc = Callable[[NominatimAPIAsync, ASGIAdaptor], Any]