WordInfos as WordInfos,
DetailedResult as DetailedResult,
ReverseResult as ReverseResult,
- ReverseResults as ReverseResults)
+ ReverseResults as ReverseResults,
+ SearchResult as SearchResult,
+ SearchResults as SearchResults)
from .localization import (Locales as Locales)
options: Mapping[str, Any]) -> str:
return format_json.format_base_json(results, options, True,
class_label='category')
+
+
+@dispatch.format_func(napi.SearchResults, 'xml')
+def _format_reverse_xml(results: napi.SearchResults, options: Mapping[str, Any]) -> str:
+ return format_xml.format_base_xml(results,
+ options, False, 'searchresults',
+ {'querystring': 'TODO'})
+
+
+@dispatch.format_func(napi.SearchResults, 'geojson')
+def _format_reverse_geojson(results: napi.SearchResults,
+ options: Mapping[str, Any]) -> str:
+ return format_json.format_base_geojson(results, options, False)
+
+
+@dispatch.format_func(napi.SearchResults, 'geocodejson')
+def _format_reverse_geocodejson(results: napi.SearchResults,
+ options: Mapping[str, Any]) -> str:
+ return format_json.format_base_geocodejson(results, options, False)
+
+
+@dispatch.format_func(napi.SearchResults, 'json')
+def _format_reverse_json(results: napi.SearchResults,
+ options: Mapping[str, Any]) -> str:
+ return format_json.format_base_json(results, options, False,
+ class_label='class')
+
+
+@dispatch.format_func(napi.SearchResults, 'jsonv2')
+def _format_reverse_jsonv2(results: napi.SearchResults,
+ options: Mapping[str, Any]) -> str:
+ return format_json.format_base_json(results, options, False,
+ class_label='category')
out.keyval('osm_key', result.category[0])\
.keyval('osm_value', result.category[1])\
.keyval('type', GEOCODEJSON_RANKS[max(3, min(28, result.rank_address))])\
- .keyval_not_none('accuracy', result.distance, transform=int)\
+ .keyval_not_none('accuracy', getattr(result, 'distance', None), transform=int)\
.keyval('label', ', '.join(label_parts))\
.keyval_not_none('name', result.names, transform=locales.display_name)\
return fmt
+ def parse_geometry_details(self, fmt: str) -> napi.LookupDetails:
+ details = napi.LookupDetails(address_details=True,
+ geometry_simplification=self.get_float('polygon_threshold', 0.0))
+ numgeoms = 0
+ if self.get_bool('polygon_geojson', False):
+ details.geometry_output |= napi.GeometryFormat.GEOJSON
+ numgeoms += 1
+ if fmt not in ('geojson', 'geocodejson'):
+ if self.get_bool('polygon_text', False):
+ details.geometry_output |= napi.GeometryFormat.TEXT
+ numgeoms += 1
+ if self.get_bool('polygon_kml', False):
+ details.geometry_output |= napi.GeometryFormat.KML
+ numgeoms += 1
+ if self.get_bool('polygon_svg', False):
+ details.geometry_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 details
+
+
async def status_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
""" Server glue for /status endpoint. See API docs for details.
"""
debug = params.setup_debugging()
coord = napi.Point(params.get_float('lon'), params.get_float('lat'))
locales = napi.Locales.from_accept_languages(params.get_accepted_languages())
+ details = params.parse_geometry_details(fmt)
zoom = max(0, min(18, params.get_int('zoom', 18)))
- details = napi.LookupDetails(address_details=True,
- geometry_simplification=params.get_float('polygon_threshold', 0.0))
- numgeoms = 0
- if params.get_bool('polygon_geojson', False):
- details.geometry_output |= napi.GeometryFormat.GEOJSON
- numgeoms += 1
- if fmt not in ('geojson', 'geocodejson'):
- if params.get_bool('polygon_text', False):
- details.geometry_output |= napi.GeometryFormat.TEXT
- numgeoms += 1
- if params.get_bool('polygon_kml', False):
- details.geometry_output |= napi.GeometryFormat.KML
- numgeoms += 1
- if params.get_bool('polygon_svg', False):
- details.geometry_output |= napi.GeometryFormat.SVG
- numgeoms += 1
-
- if numgeoms > params.config().get_int('POLYGON_OUTPUT_MAX_TYPES'):
- params.raise_error('Too many polgyon output options selected.')
result = await api.reverse(coord, REVERSE_MAX_RANKS[zoom],
params.get_layers() or
'extratags': params.get_bool('extratags', False),
'namedetails': params.get_bool('namedetails', False),
'addressdetails': params.get_bool('addressdetails', True)}
- if fmt == 'xml':
- fmt_options['xml_roottag'] = 'reversegeocode'
- fmt_options['xml_extra_info'] = {'querystring': 'TODO'}
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()
+ locales = napi.Locales.from_accept_languages(params.get_accepted_languages())
+ details = params.parse_geometry_details(fmt)
+
+ places = []
+ for oid in params.get('osm_ids', '').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 = {'locales': locales,
+ 'extratags': params.get_bool('extratags', False),
+ 'namedetails': params.get_bool('namedetails', False),
+ 'addressdetails': params.get_bool('addressdetails', True)}
+
+ output = formatting.format_result(results, fmt, fmt_options)
+
+ return params.build_response(output)
+
EndpointFunc = Callable[[napi.NominatimAPIAsync, ASGIAdaptor], Any]
REVERSE_MAX_RANKS = [2, 2, 2, # 0-2 Continent/Sea
ROUTES = [
('status', status_endpoint),
('details', details_endpoint),
- ('reverse', reverse_endpoint)
+ ('reverse', reverse_endpoint),
+ ('lookup', lookup_endpoint)
]
with pytest.raises(FakeError, match='^404 -- .*found'):
await glue.details_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+
+# lookup_endpoint()
+
+class TestLookupEndpoint:
+
+ @pytest.fixture(autouse=True)
+ def patch_lookup_func(self, monkeypatch):
+ self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0))]
+ async def _lookup(*args, **kwargs):
+ return napi.SearchResults(self.results)
+
+ monkeypatch.setattr(napi.NominatimAPIAsync, 'lookup', _lookup)
+
+
+ @pytest.mark.asyncio
+ async def test_lookup_no_params(self):
+ a = FakeAdaptor()
+ a.params['format'] = 'json'
+
+ res = await glue.lookup_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert res.output == '[]'
+
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize('param', ['w', 'bad', ''])
+ async def test_lookup_bad_params(self, param):
+ a = FakeAdaptor()
+ a.params['format'] = 'json'
+ a.params['osm_ids'] = f'W34,{param},N33333'
+
+ res = await glue.lookup_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == 1
+
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize('param', ['p234234', '4563'])
+ async def test_lookup_bad_osm_type(self, param):
+ a = FakeAdaptor()
+ a.params['format'] = 'json'
+ a.params['osm_ids'] = f'W34,{param},N33333'
+
+ res = await glue.lookup_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == 1
+
+
+ @pytest.mark.asyncio
+ async def test_lookup_working(self):
+ a = FakeAdaptor()
+ a.params['format'] = 'json'
+ a.params['osm_ids'] = 'N23,W34'
+
+ res = await glue.lookup_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == 1
projects / nominatim.git / commitdiff