nominatim/api/core.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 Implementation of classes for API access via libraries.
   9 """
  10 from typing import Mapping, Optional, Any, AsyncIterator, Dict
  11 import asyncio
  12 import contextlib
  13 from pathlib import Path
  14
  15 import sqlalchemy as sa
  16 import sqlalchemy.ext.asyncio as sa_asyncio
  17 import asyncpg
  18
  19 from nominatim.db.sqlalchemy_schema import SearchTables
  20 from nominatim.config import Configuration
  21 from nominatim.api.connection import SearchConnection
  22 from nominatim.api.status import get_status, StatusResult
  23 from nominatim.api.lookup import get_place_by_id
  24 from nominatim.api.reverse import reverse_lookup
  25 from nominatim.api.types import PlaceRef, LookupDetails, AnyPoint, DataLayer
  26 from nominatim.api.results import DetailedResult, ReverseResult
  27
  28
  29 class NominatimAPIAsync:
  30     """ API loader asynchornous version.
  31     """
  32     def __init__(self, project_dir: Path,
  33                  environ: Optional[Mapping[str, str]] = None) -> None:
  34         self.config = Configuration(project_dir, environ)
  35         self.server_version = 0
  36
  37         self._engine_lock = asyncio.Lock()
  38         self._engine: Optional[sa_asyncio.AsyncEngine] = None
  39         self._tables: Optional[SearchTables] = None
  40         self._property_cache: Dict[str, Any] = {'DB:server_version': 0}
  41
  42
  43     async def setup_database(self) -> None:
  44         """ Set up the engine and connection parameters.
  45
  46             This function will be implicitly called when the database is
  47             accessed for the first time. You may also call it explicitly to
  48             avoid that the first call is delayed by the setup.
  49         """
  50         async with self._engine_lock:
  51             if self._engine:
  52                 return
  53
  54             dsn = self.config.get_database_params()
  55
  56             dburl = sa.engine.URL.create(
  57                        'postgresql+asyncpg',
  58                        database=dsn.get('dbname'),
  59                        username=dsn.get('user'), password=dsn.get('password'),
  60                        host=dsn.get('host'), port=int(dsn['port']) if 'port' in dsn else None,
  61                        query={k: v for k, v in dsn.items()
  62                               if k not in ('user', 'password', 'dbname', 'host', 'port')})
  63             engine = sa_asyncio.create_async_engine(
  64                              dburl, future=True,
  65                              connect_args={'server_settings': {
  66                                 'DateStyle': 'sql,european',
  67                                 'max_parallel_workers_per_gather': '0'
  68                              }})
  69
  70             try:
  71                 async with engine.begin() as conn:
  72                     result = await conn.scalar(sa.text('SHOW server_version_num'))
  73                     server_version = int(result)
  74             except asyncpg.PostgresError:
  75                 server_version = 0
  76
  77             if server_version >= 110000:
  78                 @sa.event.listens_for(engine.sync_engine, "connect")
  79                 def _on_connect(dbapi_con: Any, _: Any) -> None:
  80                     cursor = dbapi_con.cursor()
  81                     cursor.execute("SET jit_above_cost TO '-1'")
  82                 # Make sure that all connections get the new settings
  83                 await self.close()
  84
  85             self._property_cache['DB:server_version'] = server_version
  86
  87             self._tables = SearchTables(sa.MetaData(), engine.name) # pylint: disable=no-member
  88             self._engine = engine
  89
  90
  91     async def close(self) -> None:
  92         """ Close all active connections to the database. The NominatimAPIAsync
  93             object remains usable after closing. If a new API functions is
  94             called, new connections are created.
  95         """
  96         if self._engine is not None:
  97             await self._engine.dispose()
  98
  99
 100     @contextlib.asynccontextmanager
 101     async def begin(self) -> AsyncIterator[SearchConnection]:
 102         """ Create a new connection with automatic transaction handling.
 103
 104             This function may be used to get low-level access to the database.
 105             Refer to the documentation of SQLAlchemy for details how to use
 106             the connection object.
 107         """
 108         if self._engine is None:
 109             await self.setup_database()
 110
 111         assert self._engine is not None
 112         assert self._tables is not None
 113
 114         async with self._engine.begin() as conn:
 115             yield SearchConnection(conn, self._tables, self._property_cache)
 116
 117
 118     async def status(self) -> StatusResult:
 119         """ Return the status of the database.
 120         """
 121         try:
 122             async with self.begin() as conn:
 123                 status = await get_status(conn)
 124         except asyncpg.PostgresError:
 125             return StatusResult(700, 'Database connection failed')
 126
 127         return status
 128
 129
 130     async def lookup(self, place: PlaceRef,
 131                      details: Optional[LookupDetails] = None) -> Optional[DetailedResult]:
 132         """ Get detailed information about a place in the database.
 133
 134             Returns None if there is no entry under the given ID.
 135         """
 136         async with self.begin() as conn:
 137             return await get_place_by_id(conn, place, details or LookupDetails())
 138
 139
 140     async def reverse(self, coord: AnyPoint, max_rank: Optional[int] = None,
 141                       layer: Optional[DataLayer] = None,
 142                       details: Optional[LookupDetails] = None) -> Optional[ReverseResult]:
 143         """ Find a place by its coordinates. Also known as reverse geocoding.
 144
 145             Returns the closest result that can be found or None if
 146             no place matches the given criteria.
 147         """
 148         # The following negation handles NaN correctly. Don't change.
 149         if not abs(coord[0]) <= 180 or not abs(coord[1]) <= 90:
 150             # There are no results to be expected outside valid coordinates.
 151             return None
 152
 153         if layer is None:
 154             layer = DataLayer.ADDRESS | DataLayer.POI
 155
 156         max_rank = max(0, min(max_rank or 30, 30))
 157
 158         async with self.begin() as conn:
 159             return await reverse_lookup(conn, coord, max_rank, layer,
 160                                         details or LookupDetails())
 161
 162
 163 class NominatimAPI:
 164     """ API loader, synchronous version.
 165     """
 166
 167     def __init__(self, project_dir: Path,
 168                  environ: Optional[Mapping[str, str]] = None) -> None:
 169         self._loop = asyncio.new_event_loop()
 170         self._async_api = NominatimAPIAsync(project_dir, environ)
 171
 172
 173     def close(self) -> None:
 174         """ Close all active connections to the database. The NominatimAPIAsync
 175             object remains usable after closing. If a new API functions is
 176             called, new connections are created.
 177         """
 178         self._loop.run_until_complete(self._async_api.close())
 179         self._loop.close()
 180
 181
 182     @property
 183     def config(self) -> Configuration:
 184         """ Return the configuration used by the API.
 185         """
 186         return self._async_api.config
 187
 188     def status(self) -> StatusResult:
 189         """ Return the status of the database.
 190         """
 191         return self._loop.run_until_complete(self._async_api.status())
 192
 193
 194     def lookup(self, place: PlaceRef,
 195                details: Optional[LookupDetails] = None) -> Optional[DetailedResult]:
 196         """ Get detailed information about a place in the database.
 197         """
 198         return self._loop.run_until_complete(self._async_api.lookup(place, details))
 199
 200
 201     def reverse(self, coord: AnyPoint, max_rank: Optional[int] = None,
 202                 layer: Optional[DataLayer] = None,
 203                 details: Optional[LookupDetails] = None) -> Optional[ReverseResult]:
 204         """ Find a place by its coordinates. Also known as reverse geocoding.
 205
 206             Returns the closest result that can be found or None if
 207             no place matches the given criteria.
 208         """
 209         return self._loop.run_until_complete(
 210                    self._async_api.reverse(coord, max_rank, layer, details))