]> git.openstreetmap.org Git - nominatim.git/blob - nominatim/api/v1/server_glue.py
factor out common server implementation code
[nominatim.git] / nominatim / api / v1 / server_glue.py
1 # SPDX-License-Identifier: GPL-2.0-only
2 #
3 # This file is part of Nominatim. (https://nominatim.org)
4 #
5 # Copyright (C) 2023 by the Nominatim developer community.
6 # For a full list of authors see the git log.
7 """
8 Generic part of the server implementation of the v1 API.
9 Combine with the scaffolding provided for the various Python ASGI frameworks.
10 """
11 from typing import Optional, Any, Type, Callable
12 import abc
13
14 import nominatim.api as napi
15 from nominatim.api.v1.format import dispatch as formatting
16
17 CONTENT_TYPE = {
18   'text': 'text/plain; charset=utf-8',
19   'xml': 'text/xml; charset=utf-8',
20   'jsonp': 'application/javascript'
21 }
22
23
24 class ASGIAdaptor(abc.ABC):
25     """ Adapter class for the different ASGI frameworks.
26         Wraps functionality over concrete requests and responses.
27     """
28
29     @abc.abstractmethod
30     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
31         """ Return an input parameter as a string. If the parameter was
32             not provided, return the 'default' value.
33         """
34
35     @abc.abstractmethod
36     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
37         """ Return a HTTP header parameter as a string. If the parameter was
38             not provided, return the 'default' value.
39         """
40
41
42     @abc.abstractmethod
43     def error(self, msg: str) -> Exception:
44         """ Construct an appropriate exception from the given error message.
45             The exception must result in a HTTP 400 error.
46         """
47
48
49     @abc.abstractmethod
50     def create_response(self, status: int, output: str, content_type: str) -> Any:
51         """ Create a response from the given parameters. The result will
52             be returned by the endpoint functions. The adaptor may also
53             return None when the response is created internally with some
54             different means.
55
56             The response must return the HTTP given status code 'status', set
57             the HTTP content-type headers to the string provided and the
58             body of the response to 'output'.
59         """
60
61
62     def build_response(self, output: str, media_type: str, status: int = 200) -> Any:
63         """ Create a response from the given output. Wraps a JSONP function
64             around the response, if necessary.
65         """
66         if media_type == 'json' and status == 200:
67             jsonp = self.get('json_callback')
68             if jsonp is not None:
69                 if any(not part.isidentifier() for part in jsonp.split('.')):
70                     raise self.error('Invalid json_callback value')
71                 output = f"{jsonp}({output})"
72                 media_type = 'jsonp'
73
74         return self.create_response(status, output,
75                                     CONTENT_TYPE.get(media_type, 'application/json'))
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             raise self.error(f"Parameter '{name}' missing.")
93
94         try:
95             return int(value)
96         except ValueError as exc:
97             raise self.error(f"Parameter '{name}' must be a number.") from exc
98
99
100     def get_bool(self, name: str, default: Optional[bool] = None) -> bool:
101         """ Return an input parameter as bool. Only '0' is accepted as
102             an input for 'false' all other inputs will be interpreted as 'true'.
103
104             If 'default' is given, then it will be returned when the parameter
105             is missing completely. When 'default' is None, an error will be
106             raised on a missing parameter.
107         """
108         value = self.get(name)
109
110         if value is None:
111             if default is not None:
112                 return default
113
114             raise self.error(f"Parameter '{name}' missing.")
115
116         return value != '0'
117
118
119 def parse_format(params: ASGIAdaptor, result_type: Type[Any], default: str) -> str:
120     """ Get and check the 'format' parameter and prepare the formatter.
121         `fmtter` is a formatter and `default` the
122         format value to assume when no parameter is present.
123     """
124     fmt = params.get('format', default=default)
125     assert fmt is not None
126
127     if not formatting.supports_format(result_type, fmt):
128         raise params.error("Parameter 'format' must be one of: " +
129                            ', '.join(formatting.list_formats(result_type)))
130
131     return fmt
132
133
134 async def status_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
135     """ Server glue for /status endpoint. See API docs for details.
136     """
137     result = await api.status()
138
139     fmt = parse_format(params, napi.StatusResult, 'text')
140
141     if fmt == 'text' and result.status:
142         status_code = 500
143     else:
144         status_code = 200
145
146     return params.build_response(formatting.format_result(result, fmt), fmt,
147                                  status=status_code)
148
149 EndpointFunc = Callable[[napi.NominatimAPIAsync, ASGIAdaptor], Any]
150
151 ROUTES = [
152     ('status', status_endpoint)
153 ]