Enhanced and refactored 'collect_os_info.py'

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

diff --git a/nominatim/db/connection.py b/nominatim/db/connection.py
index 5a1b46959d457b6ce3f9dd7e4d3e6b00590f9434..44a293d43b7f58dbfd19690525989b69333a9486 100644 (file)
--- a/nominatim/db/connection.py
+++ b/nominatim/db/connection.py
@@ -7,7 +7,7 @@
 """
 Specialised connection and cursor functions.
 """
 """
 Specialised connection and cursor functions.
 """

-from typing import List, Optional, Any, Callable, ContextManager, Dict, cast, overload, Tuple
+from typing import Optional, Any, Callable, ContextManager, Dict, cast, overload, Tuple, Iterable

 import contextlib
 import logging
 import os
 import contextlib
 import logging
 import os


@@ -36,8 +36,8 @@ class Cursor(psycopg2.extras.DictCursor):
         super().execute(query, args)
 
 
         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: Iterable[Tuple[Any, ...]],
+                       template: Optional[Query] = None) -> None:

         """ Wrapper for the psycopg2 convenience function to execute
             SQL for a list of values.
         """
         """ Wrapper for the psycopg2 convenience function to execute
             SQL for a list of values.
         """


@@ -55,7 +55,7 @@ class Cursor(psycopg2.extras.DictCursor):
         if self.rowcount != 1:
             raise RuntimeError("Query did not return a single row.")
 
         if self.rowcount != 1:
             raise RuntimeError("Query did not return a single row.")
 

-        result = self.fetchone() # type: ignore[no-untyped-call]
+        result = self.fetchone()

         assert result is not None
 
         return result[0]
         assert result is not None
 
         return result[0]


@@ -63,7 +63,7 @@ class Cursor(psycopg2.extras.DictCursor):
 
     def drop_table(self, name: str, if_exists: bool = True, cascade: bool = False) -> None:
         """ Drop the table with the given name.
 
     def drop_table(self, name: str, if_exists: bool = True, cascade: bool = False) -> None:
         """ Drop the table with the given name.

-            Set `if_exists` to False if a non-existant table should raise
+            Set `if_exists` to False if a non-existent table should raise

             an exception instead of just being ignored. If 'cascade' is set
             to True then all dependent tables are deleted as well.
         """
             an exception instead of just being ignored. If 'cascade' is set
             to True then all dependent tables are deleted as well.
         """


@@ -74,7 +74,7 @@ class Cursor(psycopg2.extras.DictCursor):
         if cascade:
             sql += ' CASCADE'
 
         if cascade:
             sql += ' CASCADE'
 

-        self.execute(pysql.SQL(sql).format(pysql.Identifier(name))) # type: ignore[no-untyped-call]
+        self.execute(pysql.SQL(sql).format(pysql.Identifier(name)))

 
 
 class Connection(psycopg2.extensions.connection):
 
 
 class Connection(psycopg2.extensions.connection):


@@ -131,7 +131,7 @@ class Connection(psycopg2.extensions.connection):
                 return False
 
             if table is not None:
                 return False
 
             if table is not None:

-                row = cur.fetchone() # type: ignore[no-untyped-call]
+                row = cur.fetchone()

                 if row is None or not isinstance(row[0], str):
                     return False
                 return row[0] == table
                 if row is None or not isinstance(row[0], str):
                     return False
                 return row[0] == table


@@ -141,7 +141,7 @@ class Connection(psycopg2.extensions.connection):
 
     def drop_table(self, name: str, if_exists: bool = True, cascade: bool = False) -> None:
         """ Drop the table with the given name.
 
     def drop_table(self, name: str, if_exists: bool = True, cascade: bool = False) -> None:
         """ Drop the table with the given name.

-            Set `if_exists` to False if a non-existant table should raise
+            Set `if_exists` to False if a non-existent table should raise

             an exception instead of just being ignored.
         """
         with self.cursor() as cur:
             an exception instead of just being ignored.
         """
         with self.cursor() as cur:


@@ -236,7 +236,7 @@ def get_pg_env(dsn: str,
     """
     env = dict(base_env if base_env is not None else os.environ)
 
     """
     env = dict(base_env if base_env is not None else os.environ)
 

-    for param, value in psycopg2.extensions.parse_dsn(dsn).items(): # type: ignore
+    for param, value in psycopg2.extensions.parse_dsn(dsn).items():

         if param in _PG_CONNECTION_STRINGS:
             env[_PG_CONNECTION_STRINGS[param]] = value
         else:
         if param in _PG_CONNECTION_STRINGS:
             env[_PG_CONNECTION_STRINGS[param]] = value
         else: