nominatim/api/connection.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   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 Extended SQLAlchemy connection class that also includes access to the schema.
   9 """
  10 from typing import cast, Any, Mapping, Sequence, Union, Dict, Optional, Set, \
  11                    Awaitable, Callable, TypeVar
  12
  13 import sqlalchemy as sa
  14 from sqlalchemy.ext.asyncio import AsyncConnection
  15
  16 from nominatim.typing import SaFromClause
  17 from nominatim.db.sqlalchemy_schema import SearchTables
  18 from nominatim.db.sqlalchemy_types import Geometry
  19 from nominatim.api.logging import log
  20
  21 T = TypeVar('T')
  22
  23 class SearchConnection:
  24     """ An extended SQLAlchemy connection class, that also contains
  25         then table definitions. The underlying asynchronous SQLAlchemy
  26         connection can be accessed with the 'connection' property.
  27         The 't' property is the collection of Nominatim tables.
  28     """
  29
  30     def __init__(self, conn: AsyncConnection,
  31                  tables: SearchTables,
  32                  properties: Dict[str, Any]) -> None:
  33         self.connection = conn
  34         self.t = tables # pylint: disable=invalid-name
  35         self._property_cache = properties
  36         self._classtables: Optional[Set[str]] = None
  37
  38
  39     async def scalar(self, sql: sa.sql.base.Executable,
  40                      params: Union[Mapping[str, Any], None] = None
  41                     ) -> Any:
  42         """ Execute a 'scalar()' query on the connection.
  43         """
  44         log().sql(self.connection, sql, params)
  45         return await self.connection.scalar(sql, params)
  46
  47
  48     async def execute(self, sql: 'sa.Executable',
  49                       params: Union[Mapping[str, Any], Sequence[Mapping[str, Any]], None] = None
  50                      ) -> 'sa.Result[Any]':
  51         """ Execute a 'execute()' query on the connection.
  52         """
  53         log().sql(self.connection, sql, params)
  54         return await self.connection.execute(sql, params)
  55
  56
  57     async def get_property(self, name: str, cached: bool = True) -> str:
  58         """ Get a property from Nominatim's property table.
  59
  60             Property values are normally cached so that they are only
  61             retrieved from the database when they are queried for the
  62             first time with this function. Set 'cached' to False to force
  63             reading the property from the database.
  64
  65             Raises a ValueError if the property does not exist.
  66         """
  67         lookup_name = f'DBPROP:{name}'
  68
  69         if cached and lookup_name in self._property_cache:
  70             return cast(str, self._property_cache[lookup_name])
  71
  72         sql = sa.select(self.t.properties.c.value)\
  73             .where(self.t.properties.c.property == name)
  74         value = await self.connection.scalar(sql)
  75
  76         if value is None:
  77             raise ValueError(f"Property '{name}' not found in database.")
  78
  79         self._property_cache[lookup_name] = cast(str, value)
  80
  81         return cast(str, value)
  82
  83
  84     async def get_db_property(self, name: str) -> Any:
  85         """ Get a setting from the database. At the moment, only
  86             'server_version', the version of the database software, can
  87             be retrieved with this function.
  88
  89             Raises a ValueError if the property does not exist.
  90         """
  91         if name != 'server_version':
  92             raise ValueError(f"DB setting '{name}' not found in database.")
  93
  94         return self._property_cache['DB:server_version']
  95
  96
  97     async def get_cached_value(self, group: str, name: str,
  98                                factory: Callable[[], Awaitable[T]]) -> T:
  99         """ Access the cache for this Nominatim instance.
 100             Each cache value needs to belong to a group and have a name.
 101             This function is for internal API use only.
 102
 103             `factory` is an async callback function that produces
 104             the value if it is not already cached.
 105
 106             Returns the cached value or the result of factory (also caching
 107             the result).
 108         """
 109         full_name = f'{group}:{name}'
 110
 111         if full_name in self._property_cache:
 112             return cast(T, self._property_cache[full_name])
 113
 114         value = await factory()
 115         self._property_cache[full_name] = value
 116
 117         return value
 118
 119
 120     async def get_class_table(self, cls: str, typ: str) -> Optional[SaFromClause]:
 121         """ Lookup up if there is a classtype table for the given category
 122             and return a SQLAlchemy table for it, if it exists.
 123         """
 124         if self._classtables is None:
 125             res = await self.execute(sa.text("""SELECT tablename FROM pg_tables
 126                                                 WHERE tablename LIKE 'place_classtype_%'
 127                                              """))
 128             self._classtables = {r[0] for r in res}
 129
 130         tablename = f"place_classtype_{cls}_{typ}"
 131
 132         if tablename not in self._classtables:
 133             return None
 134
 135         if tablename in self.t.meta.tables:
 136             return self.t.meta.tables[tablename]
 137
 138         return sa.Table(tablename, self.t.meta,
 139                         sa.Column('place_id', sa.BigInteger),
 140                         sa.Column('centroid', Geometry))