"""
Implementation of classes for API access via libraries.
"""
-from typing import Mapping, Optional, Any, AsyncIterator, Dict
+from typing import Mapping, Optional, Any, AsyncIterator, Dict, Sequence
import asyncio
import contextlib
from pathlib import Path
from nominatim.config import Configuration
from nominatim.api.connection import SearchConnection
from nominatim.api.status import get_status, StatusResult
-from nominatim.api.lookup import get_detailed_place
+from nominatim.api.lookup import get_detailed_place, get_simple_place
from nominatim.api.reverse import ReverseGeocoder
from nominatim.api.types import PlaceRef, LookupDetails, AnyPoint, DataLayer
-from nominatim.api.results import DetailedResult, ReverseResult
+from nominatim.api.results import DetailedResult, ReverseResult, SearchResults
class NominatimAPIAsync:
return status
- async def lookup(self, place: PlaceRef,
- details: Optional[LookupDetails] = None) -> Optional[DetailedResult]:
+ async def details(self, place: PlaceRef,
+ details: Optional[LookupDetails] = None) -> Optional[DetailedResult]:
""" Get detailed information about a place in the database.
Returns None if there is no entry under the given ID.
return await get_detailed_place(conn, place, details or LookupDetails())
+ async def lookup(self, places: Sequence[PlaceRef],
+ details: Optional[LookupDetails] = None) -> SearchResults:
+ """ Get simple information about a list of places.
+
+ Returns a list of place information for all IDs that were found.
+ """
+ if details is None:
+ details = LookupDetails()
+ async with self.begin() as conn:
+ return SearchResults(filter(None,
+ [await get_simple_place(conn, p, details) for p in places]))
+
+
async def reverse(self, coord: AnyPoint, max_rank: Optional[int] = None,
layer: Optional[DataLayer] = None,
details: Optional[LookupDetails] = None) -> Optional[ReverseResult]:
return self._loop.run_until_complete(self._async_api.status())
- def lookup(self, place: PlaceRef,
- details: Optional[LookupDetails] = None) -> Optional[DetailedResult]:
+ def details(self, place: PlaceRef,
+ details: Optional[LookupDetails] = None) -> Optional[DetailedResult]:
""" Get detailed information about a place in the database.
"""
- return self._loop.run_until_complete(self._async_api.lookup(place, details))
+ return self._loop.run_until_complete(self._async_api.details(place, details))
+
+
+ def lookup(self, places: Sequence[PlaceRef],
+ details: Optional[LookupDetails] = None) -> SearchResults:
+ """ Get simple information about a list of places.
+
+ Returns a list of place information for all IDs that were found.
+ """
+ return self._loop.run_until_complete(self._async_api.lookup(places, details))
def reverse(self, coord: AnyPoint, max_rank: Optional[int] = None,
indexed_date=import_date,
geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
- result = apiobj.api.lookup(idobj, napi.LookupDetails())
+ result = apiobj.api.details(idobj, napi.LookupDetails())
assert result is not None
indexed_date=import_date,
geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
- result = apiobj.api.lookup(napi.PlaceID(332), napi.LookupDetails())
+ result = apiobj.api.details(napi.PlaceID(332), napi.LookupDetails())
assert result is not None
apiobj.add_placex(place_id=332,
geometry='LINESTRING(23 34, 23.1 34)')
- result = apiobj.api.lookup(napi.PlaceID(332),
+ result = apiobj.api.details(napi.PlaceID(332),
napi.LookupDetails(geometry_output=napi.GeometryFormat.GEOJSON))
assert result.geometry == {'geojson': '{"type":"LineString","coordinates":[[23,34],[23.1,34]]}'}
country_code='pl',
rank_search=17, rank_address=16)
- result = apiobj.api.lookup(napi.PlaceID(332),
+ result = apiobj.api.details(napi.PlaceID(332),
napi.LookupDetails(address_details=True))
assert result.address_rows == [
country_code='pl', linked_place_id=45,
rank_search=27, rank_address=26)
- result = apiobj.api.lookup(napi.PlaceID(332),
+ result = apiobj.api.details(napi.PlaceID(332),
napi.LookupDetails(linked_places=True))
assert result.linked_rows == []
country_code='pl', linked_place_id=332,
rank_search=27, rank_address=26)
- result = apiobj.api.lookup(napi.PlaceID(332),
+ result = apiobj.api.details(napi.PlaceID(332),
napi.LookupDetails(linked_places=True))
assert result.linked_rows == [
country_code='pl', parent_place_id=45,
rank_search=27, rank_address=26)
- result = apiobj.api.lookup(napi.PlaceID(332),
+ result = apiobj.api.details(napi.PlaceID(332),
napi.LookupDetails(parented_places=True))
assert result.parented_rows == []
country_code='pl', parent_place_id=332,
rank_search=27, rank_address=26)
- result = apiobj.api.lookup(napi.PlaceID(332),
+ result = apiobj.api.details(napi.PlaceID(332),
napi.LookupDetails(parented_places=True))
assert result.parented_rows == [
indexed_date=import_date,
geometry='LINESTRING(23 34, 23 35)')
- result = apiobj.api.lookup(idobj, napi.LookupDetails())
+ result = apiobj.api.details(idobj, napi.LookupDetails())
assert result is not None
startnumber=11, endnumber=20, step=1)
for i in range(1, 6):
- result = apiobj.api.lookup(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
+ result = apiobj.api.details(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
assert result.place_id == 1000
for i in range(7, 11):
- result = apiobj.api.lookup(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
+ result = apiobj.api.details(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
assert result.place_id == 1001
for i in range(12, 22):
- result = apiobj.api.lookup(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
+ result = apiobj.api.details(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
assert result.place_id == 1002
country_code='pl',
rank_search=17, rank_address=16)
- result = apiobj.api.lookup(napi.PlaceID(9000),
+ result = apiobj.api.details(napi.PlaceID(9000),
napi.LookupDetails(address_details=True))
assert result.address_rows == [
osm_type='W', osm_id=6601223,
geometry='LINESTRING(23 34, 23 35)')
- result = apiobj.api.lookup(napi.PlaceID(4924), napi.LookupDetails())
+ result = apiobj.api.details(napi.PlaceID(4924), napi.LookupDetails())
assert result is not None
country_code='us',
rank_search=17, rank_address=16)
- result = apiobj.api.lookup(napi.PlaceID(9000),
+ result = apiobj.api.details(napi.PlaceID(9000),
napi.LookupDetails(address_details=True))
assert result.address_rows == [
indexed_date=import_date,
geometry='POINT(-9.45 5.6)')
- result = apiobj.api.lookup(napi.PlaceID(554), napi.LookupDetails())
+ result = apiobj.api.details(napi.PlaceID(554), napi.LookupDetails())
assert result is not None
country_code='gb',
rank_search=17, rank_address=16)
- result = apiobj.api.lookup(napi.PlaceID(9000),
+ result = apiobj.api.details(napi.PlaceID(9000),
napi.LookupDetails(address_details=True))
assert result.address_rows == [
apiobj.add_placex(place_id=1, osm_type='N', osm_id=55,
class_='place', type='suburb')
- assert apiobj.api.lookup(objid, napi.LookupDetails()) is None
+ assert apiobj.api.details(objid, napi.LookupDetails()) is None
@pytest.mark.parametrize('gtype', (napi.GeometryFormat.KML,
apiobj.add_placex(place_id=332)
with pytest.raises(ValueError):
- apiobj.api.lookup(napi.PlaceID(332),
+ apiobj.api.details(napi.PlaceID(332),
napi.LookupDetails(geometry_output=gtype))