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) 2025 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 ..types import QueryStatistics
  17 from ..result_formatting import FormatDispatcher
  18 from .content_types import CONTENT_TEXT
  19
  20
  21 class ASGIAdaptor(abc.ABC):
  22     """ Adapter class for the different ASGI frameworks.
  23         Wraps functionality over concrete requests and responses.
  24     """
  25     content_type: str = CONTENT_TEXT
  26
  27     @abc.abstractmethod
  28     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  29         """ Return an input parameter as a string. If the parameter was
  30             not provided, return the 'default' value.
  31         """
  32
  33     @abc.abstractmethod
  34     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  35         """ Return a HTTP header parameter as a string. If the parameter was
  36             not provided, return the 'default' value.
  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     @abc.abstractmethod
  46     def create_response(self, status: int, output: str, num_results: int) -> Any:
  47         """ Create a response from the given parameters. The result will
  48             be returned by the endpoint functions. The adaptor may also
  49             return None when the response is created internally with some
  50             different means.
  51
  52             The response must return the HTTP given status code 'status', set
  53             the HTTP content-type headers to the string provided and the
  54             body of the response to 'output'.
  55         """
  56
  57     @abc.abstractmethod
  58     def base_uri(self) -> str:
  59         """ Return the URI of the original request.
  60         """
  61
  62     @abc.abstractmethod
  63     def config(self) -> Configuration:
  64         """ Return the current configuration object.
  65         """
  66
  67     @abc.abstractmethod
  68     def formatting(self) -> FormatDispatcher:
  69         """ Return the formatting object to use.
  70         """
  71
  72     @abc.abstractmethod
  73     def query_stats(self) -> Optional[QueryStatistics]:
  74         """ Return the object for saving query statistics or None if
  75             no statistics are required.
  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             self.raise_error(f"Parameter '{name}' missing.")
  93
  94         try:
  95             intval = int(value)
  96         except ValueError:
  97             self.raise_error(f"Parameter '{name}' must be a number.")
  98
  99         return intval
 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     def get_bool(self, name: str, default: Optional[bool] = None) -> bool:
 128         """ Return an input parameter as bool. Only '0' is accepted as
 129             an input for 'false' all other inputs will be interpreted as 'true'.
 130
 131             If 'default' is given, then it will be returned when the parameter
 132             is missing completely. When 'default' is None, an error will be
 133             raised on a missing parameter.
 134         """
 135         value = self.get(name)
 136
 137         if value is None:
 138             if default is not None:
 139                 return default
 140
 141             self.raise_error(f"Parameter '{name}' missing.")
 142
 143         return value != '0'
 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         raise self.error(self.formatting().format_error(self.content_type, msg, status),
 151                          status)
 152
 153
 154 EndpointFunc = Callable[[NominatimAPIAsync, ASGIAdaptor], Any]