Merge remote-tracking branch 'upstream/master'

[nominatim.git] / nominatim / api / connection.py

diff --git a/nominatim/api/connection.py b/nominatim/api/connection.py
index 72cabf7814bd4d5402d4b99dada5ce720fd8c852..405213e97659d32fb9ff9d56c2478219690af6a4 100644 (file)
--- a/nominatim/api/connection.py
+++ b/nominatim/api/connection.py
@@ -7,7 +7,9 @@
 """
 Extended SQLAlchemy connection class that also includes access to the schema.
 """
 """
 Extended SQLAlchemy connection class that also includes access to the schema.
 """

-from typing import cast, Any, Mapping, Sequence, Union, Dict, Optional, Set
+from typing import cast, Any, Mapping, Sequence, Union, Dict, Optional, Set, \
+                   Awaitable, Callable, TypeVar
+import asyncio

 
 import sqlalchemy as sa
 from sqlalchemy.ext.asyncio import AsyncConnection
 
 import sqlalchemy as sa
 from sqlalchemy.ext.asyncio import AsyncConnection


@@ -17,6 +19,8 @@ from nominatim.db.sqlalchemy_schema import SearchTables
 from nominatim.db.sqlalchemy_types import Geometry
 from nominatim.api.logging import log
 
 from nominatim.db.sqlalchemy_types import Geometry
 from nominatim.api.logging import log
 

+T = TypeVar('T')
+

 class SearchConnection:
     """ An extended SQLAlchemy connection class, that also contains
         then table definitions. The underlying asynchronous SQLAlchemy
 class SearchConnection:
     """ An extended SQLAlchemy connection class, that also contains
         then table definitions. The underlying asynchronous SQLAlchemy


@@ -31,6 +35,14 @@ class SearchConnection:
         self.t = tables # pylint: disable=invalid-name
         self._property_cache = properties
         self._classtables: Optional[Set[str]] = None
         self.t = tables # pylint: disable=invalid-name
         self._property_cache = properties
         self._classtables: Optional[Set[str]] = None

+        self.query_timeout: Optional[int] = None
+
+
+    def set_query_timeout(self, timeout: Optional[int]) -> None:
+        """ Set the timeout after which a query over this connection
+            is cancelled.
+        """
+        self.query_timeout = timeout

 
 
     async def scalar(self, sql: sa.sql.base.Executable,
 
 
     async def scalar(self, sql: sa.sql.base.Executable,


@@ -39,7 +51,7 @@ class SearchConnection:
         """ Execute a 'scalar()' query on the connection.
         """
         log().sql(self.connection, sql, params)
         """ Execute a 'scalar()' query on the connection.
         """
         log().sql(self.connection, sql, params)

-        return await self.connection.scalar(sql, params)
+        return await asyncio.wait_for(self.connection.scalar(sql, params), self.query_timeout)

 
 
     async def execute(self, sql: 'sa.Executable',
 
 
     async def execute(self, sql: 'sa.Executable',


@@ -48,7 +60,7 @@ class SearchConnection:
         """ Execute a 'execute()' query on the connection.
         """
         log().sql(self.connection, sql, params)
         """ Execute a 'execute()' query on the connection.
         """
         log().sql(self.connection, sql, params)

-        return await self.connection.execute(sql, params)
+        return await asyncio.wait_for(self.connection.execute(sql, params), self.query_timeout)

 
 
     async def get_property(self, name: str, cached: bool = True) -> str:
 
 
     async def get_property(self, name: str, cached: bool = True) -> str:


@@ -61,11 +73,10 @@ class SearchConnection:
 
             Raises a ValueError if the property does not exist.
         """
 
             Raises a ValueError if the property does not exist.
         """

-        if name.startswith('DB:'):
-            raise ValueError(f"Illegal property value '{name}'.")
+        lookup_name = f'DBPROP:{name}'

 
 

-        if cached and name in self._property_cache:
-            return cast(str, self._property_cache[name])
+        if cached and lookup_name in self._property_cache:
+            return cast(str, self._property_cache[lookup_name])

 
         sql = sa.select(self.t.properties.c.value)\
             .where(self.t.properties.c.property == name)
 
         sql = sa.select(self.t.properties.c.value)\
             .where(self.t.properties.c.property == name)


@@ -74,7 +85,7 @@ class SearchConnection:
         if value is None:
             raise ValueError(f"Property '{name}' not found in database.")
 
         if value is None:
             raise ValueError(f"Property '{name}' not found in database.")
 

-        self._property_cache[name] = cast(str, value)
+        self._property_cache[lookup_name] = cast(str, value)

 
         return cast(str, value)
 
 
         return cast(str, value)
 


@@ -92,6 +103,29 @@ class SearchConnection:
         return self._property_cache['DB:server_version']
 
 
         return self._property_cache['DB:server_version']
 
 

+    async def get_cached_value(self, group: str, name: str,
+                               factory: Callable[[], Awaitable[T]]) -> T:
+        """ Access the cache for this Nominatim instance.
+            Each cache value needs to belong to a group and have a name.
+            This function is for internal API use only.
+
+            `factory` is an async callback function that produces
+            the value if it is not already cached.
+
+            Returns the cached value or the result of factory (also caching
+            the result).
+        """
+        full_name = f'{group}:{name}'
+
+        if full_name in self._property_cache:
+            return cast(T, self._property_cache[full_name])
+
+        value = await factory()
+        self._property_cache[full_name] = value
+
+        return value
+
+

     async def get_class_table(self, cls: str, typ: str) -> Optional[SaFromClause]:
         """ Lookup up if there is a classtype table for the given category
             and return a SQLAlchemy table for it, if it exists.
     async def get_class_table(self, cls: str, typ: str) -> Optional[SaFromClause]:
         """ Lookup up if there is a classtype table for the given category
             and return a SQLAlchemy table for it, if it exists.