X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/0c65289a80fb376b9d7e53653e16cef6a8fa781e..5360384b7024fb2a1f84fee55511beef41c3fd46:/nominatim/api/core.py diff --git a/nominatim/api/core.py b/nominatim/api/core.py index 415cd0aa..0f1dd715 100644 --- a/nominatim/api/core.py +++ b/nominatim/api/core.py @@ -7,33 +7,43 @@ """ 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, List, Tuple import asyncio +import sys import contextlib from pathlib import Path import sqlalchemy as sa import sqlalchemy.ext.asyncio as sa_asyncio -import asyncpg +from nominatim.errors import UsageError from nominatim.db.sqlalchemy_schema import SearchTables +from nominatim.db.async_core_library import PGCORE_LIB, PGCORE_ERROR 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_place_by_id -from nominatim.api.types import PlaceRef, LookupDetails -from nominatim.api.results import SearchResult +from nominatim.api.lookup import get_detailed_place, get_simple_place +from nominatim.api.reverse import ReverseGeocoder +from nominatim.api.search import ForwardGeocoder, Phrase, PhraseType, make_query_analyzer +import nominatim.api.types as ntyp +from nominatim.api.results import DetailedResult, ReverseResult, SearchResults class NominatimAPIAsync: """ API loader asynchornous version. """ def __init__(self, project_dir: Path, - environ: Optional[Mapping[str, str]] = None) -> None: + environ: Optional[Mapping[str, str]] = None, + loop: Optional[asyncio.AbstractEventLoop] = None) -> None: self.config = Configuration(project_dir, environ) + self.query_timeout = self.config.get_int('QUERY_TIMEOUT') \ + if self.config.QUERY_TIMEOUT else None self.server_version = 0 - self._engine_lock = asyncio.Lock() + if sys.version_info >= (3, 10): + self._engine_lock = asyncio.Lock() + else: + self._engine_lock = asyncio.Lock(loop=loop) # pylint: disable=unexpected-keyword-arg self._engine: Optional[sa_asyncio.AsyncEngine] = None self._tables: Optional[SearchTables] = None self._property_cache: Dict[str, Any] = {'DB:server_version': 0} @@ -51,26 +61,26 @@ class NominatimAPIAsync: return dsn = self.config.get_database_params() + pool_size = self.config.get_int('API_POOL_SIZE') + + query = {k: v for k, v in dsn.items() + if k not in ('user', 'password', 'dbname', 'host', 'port')} dburl = sa.engine.URL.create( - 'postgresql+asyncpg', + f'postgresql+{PGCORE_LIB}', database=dsn.get('dbname'), username=dsn.get('user'), password=dsn.get('password'), host=dsn.get('host'), port=int(dsn['port']) if 'port' in dsn else None, - query={k: v for k, v in dsn.items() - if k not in ('user', 'password', 'dbname', 'host', 'port')}) - engine = sa_asyncio.create_async_engine( - dburl, future=True, - connect_args={'server_settings': { - 'DateStyle': 'sql,european', - 'max_parallel_workers_per_gather': '0' - }}) + query=query) + engine = sa_asyncio.create_async_engine(dburl, future=True, + max_overflow=0, pool_size=pool_size, + echo=self.config.get_bool('DEBUG_SQL')) try: async with engine.begin() as conn: result = await conn.scalar(sa.text('SHOW server_version_num')) server_version = int(result) - except asyncpg.PostgresError: + except (PGCORE_ERROR, sa.exc.OperationalError): server_version = 0 if server_version >= 110000: @@ -78,6 +88,7 @@ class NominatimAPIAsync: def _on_connect(dbapi_con: Any, _: Any) -> None: cursor = dbapi_con.cursor() cursor.execute("SET jit_above_cost TO '-1'") + cursor.execute("SET max_parallel_workers_per_gather TO '0'") # Make sure that all connections get the new settings await self.close() @@ -119,21 +130,161 @@ class NominatimAPIAsync: """ try: async with self.begin() as conn: + conn.set_query_timeout(self.query_timeout) status = await get_status(conn) - except asyncpg.PostgresError: + except (PGCORE_ERROR, sa.exc.OperationalError): return StatusResult(700, 'Database connection failed') return status - async def lookup(self, place: PlaceRef, - details: LookupDetails) -> Optional[SearchResult]: + async def details(self, place: ntyp.PlaceRef, **params: Any) -> Optional[DetailedResult]: """ Get detailed information about a place in the database. Returns None if there is no entry under the given ID. """ - async with self.begin() as db: - return await get_place_by_id(db, place, details) + details = ntyp.LookupDetails.from_kwargs(params) + async with self.begin() as conn: + conn.set_query_timeout(self.query_timeout) + if details.keywords: + await make_query_analyzer(conn) + return await get_detailed_place(conn, place, details) + + + async def lookup(self, places: Sequence[ntyp.PlaceRef], **params: Any) -> SearchResults: + """ Get simple information about a list of places. + + Returns a list of place information for all IDs that were found. + """ + details = ntyp.LookupDetails.from_kwargs(params) + async with self.begin() as conn: + conn.set_query_timeout(self.query_timeout) + if details.keywords: + await make_query_analyzer(conn) + return SearchResults(filter(None, + [await get_simple_place(conn, p, details) for p in places])) + + + async def reverse(self, coord: ntyp.AnyPoint, **params: Any) -> Optional[ReverseResult]: + """ Find a place by its coordinates. Also known as reverse geocoding. + + Returns the closest result that can be found or None if + no place matches the given criteria. + """ + # The following negation handles NaN correctly. Don't change. + if not abs(coord[0]) <= 180 or not abs(coord[1]) <= 90: + # There are no results to be expected outside valid coordinates. + return None + + details = ntyp.ReverseDetails.from_kwargs(params) + async with self.begin() as conn: + conn.set_query_timeout(self.query_timeout) + if details.keywords: + await make_query_analyzer(conn) + geocoder = ReverseGeocoder(conn, details) + return await geocoder.lookup(coord) + + + async def search(self, query: str, **params: Any) -> SearchResults: + """ Find a place by free-text search. Also known as forward geocoding. + """ + query = query.strip() + if not query: + raise UsageError('Nothing to search for.') + + async with self.begin() as conn: + conn.set_query_timeout(self.query_timeout) + geocoder = ForwardGeocoder(conn, ntyp.SearchDetails.from_kwargs(params), + self.config.get_int('REQUEST_TIMEOUT') \ + if self.config.REQUEST_TIMEOUT else None) + phrases = [Phrase(PhraseType.NONE, p.strip()) for p in query.split(',')] + return await geocoder.lookup(phrases) + + + # pylint: disable=too-many-arguments,too-many-branches + async def search_address(self, amenity: Optional[str] = None, + street: Optional[str] = None, + city: Optional[str] = None, + county: Optional[str] = None, + state: Optional[str] = None, + country: Optional[str] = None, + postalcode: Optional[str] = None, + **params: Any) -> SearchResults: + """ Find an address using structured search. + """ + async with self.begin() as conn: + conn.set_query_timeout(self.query_timeout) + details = ntyp.SearchDetails.from_kwargs(params) + + phrases: List[Phrase] = [] + + if amenity: + phrases.append(Phrase(PhraseType.AMENITY, amenity)) + if street: + phrases.append(Phrase(PhraseType.STREET, street)) + if city: + phrases.append(Phrase(PhraseType.CITY, city)) + if county: + phrases.append(Phrase(PhraseType.COUNTY, county)) + if state: + phrases.append(Phrase(PhraseType.STATE, state)) + if postalcode: + phrases.append(Phrase(PhraseType.POSTCODE, postalcode)) + if country: + phrases.append(Phrase(PhraseType.COUNTRY, country)) + + if not phrases: + raise UsageError('Nothing to search for.') + + if amenity or street: + details.restrict_min_max_rank(26, 30) + elif city: + details.restrict_min_max_rank(13, 25) + elif county: + details.restrict_min_max_rank(10, 12) + elif state: + details.restrict_min_max_rank(5, 9) + elif postalcode: + details.restrict_min_max_rank(5, 11) + else: + details.restrict_min_max_rank(4, 4) + + if 'layers' not in params: + details.layers = ntyp.DataLayer.ADDRESS + if amenity: + details.layers |= ntyp.DataLayer.POI + + geocoder = ForwardGeocoder(conn, details, + self.config.get_int('REQUEST_TIMEOUT') \ + if self.config.REQUEST_TIMEOUT else None) + return await geocoder.lookup(phrases) + + + async def search_category(self, categories: List[Tuple[str, str]], + near_query: Optional[str] = None, + **params: Any) -> SearchResults: + """ Find an object of a certain category near another place. + The near place may either be given as an unstructured search + query in itself or as coordinates. + """ + if not categories: + return SearchResults() + + details = ntyp.SearchDetails.from_kwargs(params) + async with self.begin() as conn: + conn.set_query_timeout(self.query_timeout) + if near_query: + phrases = [Phrase(PhraseType.NONE, p) for p in near_query.split(',')] + else: + phrases = [] + if details.keywords: + await make_query_analyzer(conn) + + geocoder = ForwardGeocoder(conn, details, + self.config.get_int('REQUEST_TIMEOUT') \ + if self.config.REQUEST_TIMEOUT else None) + return await geocoder.lookup_pois(categories, phrases) + class NominatimAPI: @@ -143,7 +294,7 @@ class NominatimAPI: def __init__(self, project_dir: Path, environ: Optional[Mapping[str, str]] = None) -> None: self._loop = asyncio.new_event_loop() - self._async_api = NominatimAPIAsync(project_dir, environ) + self._async_api = NominatimAPIAsync(project_dir, environ, loop=self._loop) def close(self) -> None: @@ -167,8 +318,59 @@ class NominatimAPI: return self._loop.run_until_complete(self._async_api.status()) - def lookup(self, place: PlaceRef, - details: LookupDetails) -> Optional[SearchResult]: + def details(self, place: ntyp.PlaceRef, **params: Any) -> 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, **params)) + + + def lookup(self, places: Sequence[ntyp.PlaceRef], **params: Any) -> 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, **params)) + + + def reverse(self, coord: ntyp.AnyPoint, **params: Any) -> Optional[ReverseResult]: + """ Find a place by its coordinates. Also known as reverse geocoding. + + Returns the closest result that can be found or None if + no place matches the given criteria. + """ + return self._loop.run_until_complete(self._async_api.reverse(coord, **params)) + + + def search(self, query: str, **params: Any) -> SearchResults: + """ Find a place by free-text search. Also known as forward geocoding. + """ + return self._loop.run_until_complete( + self._async_api.search(query, **params)) + + + # pylint: disable=too-many-arguments + def search_address(self, amenity: Optional[str] = None, + street: Optional[str] = None, + city: Optional[str] = None, + county: Optional[str] = None, + state: Optional[str] = None, + country: Optional[str] = None, + postalcode: Optional[str] = None, + **params: Any) -> SearchResults: + """ Find an address using structured search. + """ + return self._loop.run_until_complete( + self._async_api.search_address(amenity, street, city, county, + state, country, postalcode, **params)) + + + def search_category(self, categories: List[Tuple[str, str]], + near_query: Optional[str] = None, + **params: Any) -> SearchResults: + """ Find an object of a certain category near another place. + The near place may either be given as an unstructured search + query in itself or as a geographic area through the + viewbox or near parameters. + """ + return self._loop.run_until_complete( + self._async_api.search_category(categories, near_query, **params))