Generic part of the server implementation of the v1 API.
Combine with the scaffolding provided for the various Python ASGI frameworks.
"""
-from typing import Optional, Any, Type, Callable, NoReturn
+from typing import Optional, Any, Type, Callable, NoReturn, Dict, cast
+from functools import reduce
import abc
+import dataclasses
+import math
+from urllib.parse import urlencode
+from nominatim.errors import UsageError
from nominatim.config import Configuration
import nominatim.api as napi
import nominatim.api.logging as loglib
from nominatim.api.v1.format import dispatch as formatting
+from nominatim.api.v1 import helpers
CONTENT_TYPE = {
'text': 'text/plain; charset=utf-8',
'debug': 'text/html; charset=utf-8'
}
-
class ASGIAdaptor(abc.ABC):
""" Adapter class for the different ASGI frameworks.
Wraps functionality over concrete requests and responses.
return intval
+
+ def get_float(self, name: str, default: Optional[float] = None) -> float:
+ """ Return an input parameter as a flaoting-point number. Raises an
+ exception if the parameter is given but not in an float format.
+
+ If 'default' is given, then it will be returned when the parameter
+ is missing completely. When 'default' is None, an error will be
+ raised on a missing parameter.
+ """
+ value = self.get(name)
+
+ if value is None:
+ if default is not None:
+ return default
+
+ self.raise_error(f"Parameter '{name}' missing.")
+
+ try:
+ fval = float(value)
+ except ValueError:
+ self.raise_error(f"Parameter '{name}' must be a number.")
+
+ if math.isnan(fval) or math.isinf(fval):
+ self.raise_error(f"Parameter '{name}' must be a number.")
+
+ return fval
+
+
def get_bool(self, name: str, default: Optional[bool] = None) -> bool:
""" Return an input parameter as bool. Only '0' is accepted as
an input for 'false' all other inputs will be interpreted as 'true'.
return False
+ def get_layers(self) -> Optional[napi.DataLayer]:
+ """ Return a parsed version of the layer parameter.
+ """
+ param = self.get('layer', None)
+ if param is None:
+ return None
+
+ return cast(napi.DataLayer,
+ reduce(napi.DataLayer.__or__,
+ (getattr(napi.DataLayer, s.upper()) for s in param.split(','))))
+
+
def parse_format(self, result_type: Type[Any], default: str) -> str:
""" Get and check the 'format' parameter and prepare the formatter.
`result_type` is the type of result to be returned by the function
return fmt
+ def parse_geometry_details(self, fmt: str) -> Dict[str, Any]:
+ """ Create details strucutre from the supplied geometry parameters.
+ """
+ numgeoms = 0
+ output = napi.GeometryFormat.NONE
+ if self.get_bool('polygon_geojson', False):
+ output |= napi.GeometryFormat.GEOJSON
+ numgeoms += 1
+ if fmt not in ('geojson', 'geocodejson'):
+ if self.get_bool('polygon_text', False):
+ output |= napi.GeometryFormat.TEXT
+ numgeoms += 1
+ if self.get_bool('polygon_kml', False):
+ output |= napi.GeometryFormat.KML
+ numgeoms += 1
+ if self.get_bool('polygon_svg', False):
+ output |= napi.GeometryFormat.SVG
+ numgeoms += 1
+
+ if numgeoms > self.config().get_int('POLYGON_OUTPUT_MAX_TYPES'):
+ self.raise_error('Too many polgyon output options selected.')
+
+ return {'address_details': True,
+ 'geometry_simplification': self.get_float('polygon_threshold', 0.0),
+ 'geometry_output': output
+ }
+
+
async def status_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
""" Server glue for /status endpoint. See API docs for details.
"""
debug = params.setup_debugging()
- details = napi.LookupDetails(address_details=params.get_bool('addressdetails', False),
- linked_places=params.get_bool('linkedplaces', False),
- parented_places=params.get_bool('hierarchy', False),
- keywords=params.get_bool('keywords', False))
-
- if params.get_bool('polygon_geojson', False):
- details.geometry_output = napi.GeometryFormat.GEOJSON
-
locales = napi.Locales.from_accept_languages(params.get_accepted_languages())
- result = await api.lookup(place, details)
+ result = await api.details(place,
+ address_details=params.get_bool('addressdetails', False),
+ linked_places=params.get_bool('linkedplaces', False),
+ parented_places=params.get_bool('hierarchy', False),
+ keywords=params.get_bool('keywords', False),
+ geometry_output = napi.GeometryFormat.GEOJSON
+ if params.get_bool('polygon_geojson', False)
+ else napi.GeometryFormat.NONE
+ )
if debug:
return params.build_response(loglib.get_and_disable())
if result is None:
params.raise_error('No place with that OSM ID found.', status=404)
+ result.localize(locales)
+
output = formatting.format_result(result, fmt,
{'locales': locales,
'group_hierarchy': params.get_bool('group_hierarchy', False),
return params.build_response(output)
+async def reverse_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
+ """ Server glue for /reverse endpoint. See API docs for details.
+ """
+ fmt = params.parse_format(napi.ReverseResults, 'xml')
+ debug = params.setup_debugging()
+ coord = napi.Point(params.get_float('lon'), params.get_float('lat'))
+
+ details = params.parse_geometry_details(fmt)
+ details['max_rank'] = helpers.zoom_to_rank(params.get_int('zoom', 18))
+ details['layers'] = params.get_layers()
+
+ result = await api.reverse(coord, **details)
+
+ if debug:
+ return params.build_response(loglib.get_and_disable())
+
+ if fmt == 'xml':
+ queryparts = {'lat': str(coord.lat), 'lon': str(coord.lon), 'format': 'xml'}
+ zoom = params.get('zoom', None)
+ if zoom:
+ queryparts['zoom'] = zoom
+ query = urlencode(queryparts)
+ else:
+ query = ''
+
+ fmt_options = {'query': query,
+ 'extratags': params.get_bool('extratags', False),
+ 'namedetails': params.get_bool('namedetails', False),
+ 'addressdetails': params.get_bool('addressdetails', True)}
+
+ if result:
+ result.localize(napi.Locales.from_accept_languages(params.get_accepted_languages()))
+
+ output = formatting.format_result(napi.ReverseResults([result] if result else []),
+ fmt, fmt_options)
+
+ return params.build_response(output)
+
+
+async def lookup_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
+ """ Server glue for /lookup endpoint. See API docs for details.
+ """
+ fmt = params.parse_format(napi.SearchResults, 'xml')
+ debug = params.setup_debugging()
+ details = params.parse_geometry_details(fmt)
+
+ places = []
+ for oid in (params.get('osm_ids') or '').split(','):
+ oid = oid.strip()
+ if len(oid) > 1 and oid[0] in 'RNWrnw' and oid[1:].isdigit():
+ places.append(napi.OsmID(oid[0], int(oid[1:])))
+
+ if places:
+ results = await api.lookup(places, **details)
+ else:
+ results = napi.SearchResults()
+
+ if debug:
+ return params.build_response(loglib.get_and_disable())
+
+ fmt_options = {'extratags': params.get_bool('extratags', False),
+ 'namedetails': params.get_bool('namedetails', False),
+ 'addressdetails': params.get_bool('addressdetails', True)}
+
+ results.localize(napi.Locales.from_accept_languages(params.get_accepted_languages()))
+
+ output = formatting.format_result(results, fmt, fmt_options)
+
+ return params.build_response(output)
+
+
+async def _unstructured_search(query: str, api: napi.NominatimAPIAsync,
+ details: Dict[str, Any]) -> napi.SearchResults:
+ if not query:
+ return napi.SearchResults()
+
+ # Extract special format for coordinates from query.
+ query, x, y = helpers.extract_coords_from_query(query)
+ if x is not None:
+ assert y is not None
+ details['near'] = napi.Point(x, y)
+ details['near_radius'] = 0.1
+
+ # If no query is left, revert to reverse search.
+ if x is not None and not query:
+ result = await api.reverse(details['near'], **details)
+ if not result:
+ return napi.SearchResults()
+
+ return napi.SearchResults(
+ [napi.SearchResult(**{f.name: getattr(result, f.name)
+ for f in dataclasses.fields(napi.SearchResult)
+ if hasattr(result, f.name)})])
+
+ query, cls, typ = helpers.extract_category_from_query(query)
+ if cls is not None:
+ assert typ is not None
+ return await api.search_category([(cls, typ)], near_query=query, **details)
+
+ return await api.search(query, **details)
+
+
+async def search_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
+ """ Server glue for /search endpoint. See API docs for details.
+ """
+ fmt = params.parse_format(napi.SearchResults, 'jsonv2')
+ debug = params.setup_debugging()
+ details = params.parse_geometry_details(fmt)
+
+ details['countries'] = params.get('countrycodes', None)
+ details['excluded'] = params.get('exclude_place_ids', None)
+ details['viewbox'] = params.get('viewbox', None) or params.get('viewboxlbrt', None)
+ details['bounded_viewbox'] = params.get_bool('bounded', False)
+ details['dedupe'] = params.get_bool('dedupe', True)
+
+ max_results = max(1, min(50, params.get_int('limit', 10)))
+ details['max_results'] = max_results + min(10, max_results) \
+ if details['dedupe'] else max_results
+
+ details['min_rank'], details['max_rank'] = \
+ helpers.feature_type_to_rank(params.get('featureType', ''))
+
+ query = params.get('q', None)
+ queryparts = {}
+ try:
+ if query is not None:
+ queryparts['q'] = query
+ results = await _unstructured_search(query, api, details)
+ else:
+ for key in ('amenity', 'street', 'city', 'county', 'state', 'postalcode', 'country'):
+ details[key] = params.get(key, None)
+ if details[key]:
+ queryparts[key] = details[key]
+ query = ', '.join(queryparts.values())
+
+ results = await api.search_address(**details)
+ except UsageError as err:
+ params.raise_error(str(err))
+
+ results.localize(napi.Locales.from_accept_languages(params.get_accepted_languages()))
+
+ if details['dedupe'] and len(results) > 1:
+ results = helpers.deduplicate_results(results, max_results)
+
+ if debug:
+ return params.build_response(loglib.get_and_disable())
+
+ if fmt == 'xml':
+ helpers.extend_query_parts(queryparts, details,
+ params.get('featureType', ''),
+ params.get_bool('namedetails', False),
+ params.get_bool('extratags', False),
+ (str(r.place_id) for r in results if r.place_id))
+ queryparts['format'] = fmt
+
+ moreurl = urlencode(queryparts)
+ else:
+ moreurl = ''
+
+ fmt_options = {'query': query, 'more_url': moreurl,
+ 'exclude_place_ids': queryparts.get('exclude_place_ids'),
+ 'viewbox': queryparts.get('viewbox'),
+ 'extratags': params.get_bool('extratags', False),
+ 'namedetails': params.get_bool('namedetails', False),
+ 'addressdetails': params.get_bool('addressdetails', False)}
+
+ output = formatting.format_result(results, fmt, fmt_options)
+
+ return params.build_response(output)
+
+
EndpointFunc = Callable[[napi.NominatimAPIAsync, ASGIAdaptor], Any]
ROUTES = [
('status', status_endpoint),
- ('details', details_endpoint)
+ ('details', details_endpoint),
+ ('reverse', reverse_endpoint),
+ ('lookup', lookup_endpoint),
+ ('search', search_endpoint)
]
projects / nominatim.git / blobdiff