"""
Specialised connection and cursor functions.
"""
-from typing import Union, List, Optional, Any, Callable, ContextManager, Mapping, cast, TypeVar, overload, Tuple, Sequence
+from typing import List, Optional, Any, Callable, ContextManager, Mapping, cast, overload, Tuple
import contextlib
import logging
import os
import psycopg2.extras
from psycopg2 import sql as pysql
+from nominatim.typing import Query, T_cursor
from nominatim.errors import UsageError
-Query = Union[str, bytes, pysql.Composable]
-T = TypeVar('T', bound=psycopg2.extensions.cursor)
-
LOG = logging.getLogger()
class _Cursor(psycopg2.extras.DictCursor):
super().execute(query, args)
- def execute_values(self, sql: Query, argslist: List[Any], template: Optional[str] = None) -> None:
+ def execute_values(self, sql: Query, argslist: List[Any],
+ template: Optional[str] = None) -> None:
""" Wrapper for the psycopg2 convenience function to execute
SQL for a list of values.
"""
...
@overload
- def cursor(self, cursor_factory: Callable[..., T]) -> T:
+ def cursor(self, cursor_factory: Callable[..., T_cursor]) -> T_cursor:
...
def cursor(self, cursor_factory = _Cursor, **kwargs): # type: ignore
def get_pg_env(dsn: str,
- base_env: Optional[Mapping[str, Optional[str]]] = None) -> Mapping[str, Optional[str]]:
+ base_env: Optional[Mapping[str, str]] = None) -> Mapping[str, str]:
""" Return a copy of `base_env` with the environment variables for
PostgresSQL set up from the given database connection string.
If `base_env` is None, then the OS environment is used as a base