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 Any, Mapping, Sequence, Union
  11
  12 import sqlalchemy as sa
  13 from sqlalchemy.ext.asyncio import AsyncConnection
  14
  15 from nominatim.db.sqlalchemy_schema import SearchTables
  16
  17 class SearchConnection:
  18     """ An extended SQLAlchemy connection class, that also contains
  19         then table definitions. The underlying asynchronous SQLAlchemy
  20         connection can be accessed with the 'connection' property.
  21         The 't' property is the collection of Nominatim tables.
  22     """
  23
  24     def __init__(self, conn: AsyncConnection,
  25                  tables: SearchTables) -> None:
  26         self.connection = conn
  27         self.t = tables # pylint: disable=invalid-name
  28
  29
  30     async def scalar(self, sql: sa.sql.base.Executable,
  31                      params: Union[Mapping[str, Any], None] = None
  32                     ) -> Any:
  33         """ Execute a 'scalar()' query on the connection.
  34         """
  35         return await self.connection.scalar(sql, params)
  36
  37
  38     async def execute(self, sql: sa.sql.base.Executable,
  39                       params: Union[Mapping[str, Any], Sequence[Mapping[str, Any]], None] = None
  40                      ) -> sa.engine.Result:
  41         """ Execute a 'execute()' query on the connection.
  42         """
  43         return await self.connection.execute(sql, params)