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