Merge pull request #3385 from mtmail/docs-reverse-only

[nominatim.git] / nominatim / tools / freeze.py

diff --git a/nominatim/tools/freeze.py b/nominatim/tools/freeze.py
index cc1bf97e09a92556cc1cc79cd47cfe03e605181c..602def55b55ac3bb8b52212d43cab04d2361f37b 100644 (file)
--- a/nominatim/tools/freeze.py
+++ b/nominatim/tools/freeze.py
@@ -1,8 +1,19 @@
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.

 """
 Functions for removing unnecessary data from the database.
 """
 """
 Functions for removing unnecessary data from the database.
 """

+from typing import Optional

 from pathlib import Path
 
 from pathlib import Path
 

+from psycopg2 import sql as pysql
+
+from nominatim.db.connection import Connection
+

 UPDATE_TABLES = [
     'address_levels',
     'gb_postcode',
 UPDATE_TABLES = [
     'address_levels',
     'gb_postcode',


@@ -17,27 +28,31 @@ UPDATE_TABLES = [
     'wikipedia_%'
 ]
 
     'wikipedia_%'
 ]
 

-def drop_update_tables(conn):
+def drop_update_tables(conn: Connection) -> None:

     """ Drop all tables only necessary for updating the database from
         OSM replication data.
     """
     """ Drop all tables only necessary for updating the database from
         OSM replication data.
     """

-
-    where = ' or '.join(["(tablename LIKE '{}')".format(t) for t in UPDATE_TABLES])
+    parts = (pysql.SQL("(tablename LIKE {})").format(pysql.Literal(t)) for t in UPDATE_TABLES)

 
     with conn.cursor() as cur:
 
     with conn.cursor() as cur:

-        cur.execute("SELECT tablename FROM pg_tables WHERE " + where)
+        cur.execute(pysql.SQL("SELECT tablename FROM pg_tables WHERE ")
+                    + pysql.SQL(' or ').join(parts))

         tables = [r[0] for r in cur]
 
         for table in tables:
         tables = [r[0] for r in cur]
 
         for table in tables:

-            cur.execute('DROP TABLE IF EXISTS "{}" CASCADE'.format(table))
+            cur.drop_table(table, cascade=True)

 
     conn.commit()
 
 
 
     conn.commit()
 
 

-def drop_flatnode_file(fname):
+def drop_flatnode_file(fpath: Optional[Path]) -> None:

     """ Remove the flatnode file if it exists.
     """
     """ Remove the flatnode file if it exists.
     """

-    if fname:
-        fpath = Path(fname)
-        if fpath.exists():
-            fpath.unlink()
+    if fpath and fpath.exists():
+        fpath.unlink()
+
+def is_frozen(conn: Connection) -> bool:
+    """ Returns true if database is in a frozen state
+    """
+
+    return conn.table_exists('place') is False