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
  18 CONTENT_TEXT = 'text/plain; charset=utf-8'
  19 CONTENT_XML = 'text/xml; charset=utf-8'
  20 CONTENT_HTML = 'text/html; charset=utf-8'
  21 CONTENT_JSON = 'application/json; charset=utf-8'
  22
  23 CONTENT_TYPE = {'text': CONTENT_TEXT, 'xml': CONTENT_XML, 'debug': CONTENT_HTML}
  24
  25 class ASGIAdaptor(abc.ABC):
  26     """ Adapter class for the different ASGI frameworks.
  27         Wraps functionality over concrete requests and responses.
  28     """
  29     content_type: str = CONTENT_TEXT
  30
  31     @abc.abstractmethod
  32     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  33         """ Return an input parameter as a string. If the parameter was
  34             not provided, return the 'default' value.
  35         """
  36
  37     @abc.abstractmethod
  38     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  39         """ Return a HTTP header parameter as a string. If the parameter was
  40             not provided, return the 'default' value.
  41         """
  42
  43
  44     @abc.abstractmethod
  45     def error(self, msg: str, status: int = 400) -> Exception:
  46         """ Construct an appropriate exception from the given error message.
  47             The exception must result in a HTTP error with the given status.
  48         """
  49
  50
  51     @abc.abstractmethod
  52     def create_response(self, status: int, output: str, num_results: int) -> Any:
  53         """ Create a response from the given parameters. The result will
  54             be returned by the endpoint functions. The adaptor may also
  55             return None when the response is created internally with some
  56             different means.
  57
  58             The response must return the HTTP given status code 'status', set
  59             the HTTP content-type headers to the string provided and the
  60             body of the response to 'output'.
  61         """
  62
  63     @abc.abstractmethod
  64     def base_uri(self) -> str:
  65         """ Return the URI of the original request.
  66         """
  67
  68
  69     @abc.abstractmethod
  70     def config(self) -> Configuration:
  71         """ Return the current configuration object.
  72         """
  73
  74
  75     def get_int(self, name: str, default: Optional[int] = None) -> int:
  76         """ Return an input parameter as an int. Raises an exception if
  77             the parameter is given but not in an integer format.
  78
  79             If 'default' is given, then it will be returned when the parameter
  80             is missing completely. When 'default' is None, an error will be
  81             raised on a missing parameter.
  82         """
  83         value = self.get(name)
  84
  85         if value is None:
  86             if default is not None:
  87                 return default
  88
  89             self.raise_error(f"Parameter '{name}' missing.")
  90
  91         try:
  92             intval = int(value)
  93         except ValueError:
  94             self.raise_error(f"Parameter '{name}' must be a number.")
  95
  96         return intval
  97
  98
  99     def get_float(self, name: str, default: Optional[float] = None) -> float:
 100         """ Return an input parameter as a flaoting-point number. Raises an
 101             exception if the parameter is given but not in an float format.
 102
 103             If 'default' is given, then it will be returned when the parameter
 104             is missing completely. When 'default' is None, an error will be
 105             raised on a missing parameter.
 106         """
 107         value = self.get(name)
 108
 109         if value is None:
 110             if default is not None:
 111                 return default
 112
 113             self.raise_error(f"Parameter '{name}' missing.")
 114
 115         try:
 116             fval = float(value)
 117         except ValueError:
 118             self.raise_error(f"Parameter '{name}' must be a number.")
 119
 120         if math.isnan(fval) or math.isinf(fval):
 121             self.raise_error(f"Parameter '{name}' must be a number.")
 122
 123         return fval
 124
 125
 126     def get_bool(self, name: str, default: Optional[bool] = None) -> bool:
 127         """ Return an input parameter as bool. Only '0' is accepted as
 128             an input for 'false' all other inputs will be interpreted as 'true'.
 129
 130             If 'default' is given, then it will be returned when the parameter
 131             is missing completely. When 'default' is None, an error will be
 132             raised on a missing parameter.
 133         """
 134         value = self.get(name)
 135
 136         if value is None:
 137             if default is not None:
 138                 return default
 139
 140             self.raise_error(f"Parameter '{name}' missing.")
 141
 142         return value != '0'
 143
 144
 145     def raise_error(self, msg: str, status: int = 400) -> NoReturn:
 146         """ Raise an exception resulting in the given HTTP status and
 147             message. The message will be formatted according to the
 148             output format chosen by the request.
 149         """
 150         if self.content_type == CONTENT_XML:
 151             msg = f"""<?xml version="1.0" encoding="UTF-8" ?>
 152                       <error>
 153                         <code>{status}</code>
 154                         <message>{msg}</message>
 155                       </error>
 156                    """
 157         elif self.content_type == CONTENT_JSON:
 158             msg = f"""{{"error":{{"code":{status},"message":"{msg}"}}}}"""
 159         elif self.content_type == CONTENT_HTML:
 160             loglib.log().section('Execution error')
 161             loglib.log().var_dump('Status', status)
 162             loglib.log().var_dump('Message', msg)
 163             msg = loglib.get_and_disable()
 164
 165         raise self.error(msg, status)
 166
 167
 168 EndpointFunc = Callable[[NominatimAPIAsync, ASGIAdaptor], Any]