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