-# SPDX-License-Identifier: GPL-2.0-only
+# SPDX-License-Identifier: GPL-3.0-or-later
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2022 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
-Specialised psycopg2 cursor with shortcut functions useful for testing.
+Specialised psycopg cursor with shortcut functions useful for testing.
"""
-import psycopg2.extras
+import psycopg
-class CursorForTesting(psycopg2.extras.DictCursor):
+
+class CursorForTesting(psycopg.Cursor):
""" Extension to the DictCursor class that provides execution
short-cuts that simplify writing assertions.
"""
assert self.rowcount == 1
return self.fetchone()[0]
-
def row_set(self, sql, params=None):
""" Execute a query and return the result as a set of tuples.
Fails when the SQL command returns duplicate rows.
return result
-
def table_exists(self, table):
""" Check that a table with the given name exists in the database.
"""
WHERE tablename = %s""", (table, ))
return num == 1
-
def index_exists(self, table, index):
""" Check that an indexwith the given name exists on the given table.
"""
(table, index))
return num == 1
-
def table_rows(self, table, where=None):
""" Return the number of rows in the given table.
"""
return self.scalar('SELECT count(*) FROM ' + table)
return self.scalar('SELECT count(*) FROM {} WHERE {}'.format(table, where))
-
-
- def execute_values(self, *args, **kwargs):
- """ Execute the execute_values() function on the cursor.
- """
- psycopg2.extras.execute_values(self, *args, **kwargs)