adapt to new version of pylint

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

diff --git a/nominatim/db/status.py b/nominatim/db/status.py
index e63a40f9ba67a6ec81ec43003c4ce1984ab6d9e1..2c01de71466dca8fb237461509062c5c8c3e248a 100644 (file)
--- a/nominatim/db/status.py
+++ b/nominatim/db/status.py
@@ -1,22 +1,44 @@
+# 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.
 """
 Access and helper functions for the status and status log table.
 """
+from typing import Optional, Tuple, cast
 import datetime as dt
 import logging
 import re
 
+from nominatim.db.connection import Connection
 from nominatim.tools.exec_utils import get_url
 from nominatim.errors import UsageError
+from nominatim.typing import TypedDict
 
 LOG = logging.getLogger()
+ISODATE_FORMAT = '%Y-%m-%dT%H:%M:%S'
 
-def compute_database_date(conn):
+
+class StatusRow(TypedDict):
+    """ Dictionary of columns of the import_status table.
+    """
+    lastimportdate: dt.datetime
+    sequence_id: Optional[int]
+    indexed: Optional[bool]
+
+
+def compute_database_date(conn: Connection) -> dt.datetime:
     """ Determine the date of the database from the newest object in the
         data base.
     """
     # First, find the node with the highest ID in the database
     with conn.cursor() as cur:
-        osmid = cur.scalar("SELECT max(osm_id) FROM place WHERE osm_type='N'")
+        if conn.table_exists('place'):
+            osmid = cur.scalar("SELECT max(osm_id) FROM place WHERE osm_type='N'")
+        else:
+            osmid = cur.scalar("SELECT max(osm_id) FROM placex WHERE osm_type='N'")
 
         if osmid is None:
             LOG.fatal("No data found in the database.")
@@ -24,7 +46,7 @@ def compute_database_date(conn):
 
     LOG.info("Using node id %d for timestamp lookup", osmid)
     # Get the node from the API to find the timestamp when it was created.
-    node_url = 'https://www.openstreetmap.org/api/0.6/node/{}/1'.format(osmid)
+    node_url = f'https://www.openstreetmap.org/api/0.6/node/{osmid}/1'
     data = get_url(node_url)
 
     match = re.search(r'timestamp="((\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2}))Z"', data)
@@ -34,15 +56,17 @@ def compute_database_date(conn):
                   "URL used: %s", node_url)
         raise UsageError("Bad API data.")
 
-    LOG.debug("Found timestamp %s", match[1])
+    LOG.debug("Found timestamp %s", match.group(1))
 
-    return dt.datetime.fromisoformat(match[1]).replace(tzinfo=dt.timezone.utc)
+    return dt.datetime.strptime(match.group(1), ISODATE_FORMAT).replace(tzinfo=dt.timezone.utc)
 
 
-def set_status(conn, date, seq=None, indexed=True):
+def set_status(conn: Connection, date: Optional[dt.datetime],
+               seq: Optional[int] = None, indexed: bool = True) -> None:
     """ Replace the current status with the given status. If date is `None`
         then only sequence and indexed will be updated as given. Otherwise
         the whole status is replaced.
+        The change will be committed to the database.
     """
     assert date is None or date.tzinfo == dt.timezone.utc
     with conn.cursor() as cur:
@@ -57,7 +81,7 @@ def set_status(conn, date, seq=None, indexed=True):
     conn.commit()
 
 
-def get_status(conn):
+def get_status(conn: Connection) -> Tuple[Optional[dt.datetime], Optional[int], Optional[bool]]:
     """ Return the current status as a triple of (date, sequence, indexed).
         If status has not been set up yet, a triple of None is returned.
     """
@@ -66,11 +90,11 @@ def get_status(conn):
         if cur.rowcount < 1:
             return None, None, None
 
-        row = cur.fetchone()
+        row = cast(StatusRow, cur.fetchone())
         return row['lastimportdate'], row['sequence_id'], row['indexed']
 
 
-def set_indexed(conn, state):
+def set_indexed(conn: Connection, state: bool) -> None:
     """ Set the indexed flag in the status table to the given state.
     """
     with conn.cursor() as cur:
@@ -78,7 +102,8 @@ def set_indexed(conn, state):
     conn.commit()
 
 
-def log_status(conn, start, event, batchsize=None):
+def log_status(conn: Connection, start: dt.datetime,
+               event: str, batchsize: Optional[int] = None) -> None:
     """ Write a new status line to the `import_osmosis_log` table.
     """
     with conn.cursor() as cur:
@@ -86,3 +111,4 @@ def log_status(conn, start, event, batchsize=None):
                        (batchend, batchseq, batchsize, starttime, endtime, event)
                        SELECT lastimportdate, sequence_id, %s, %s, now(), %s FROM import_status""",
                     (batchsize, start, event))
+    conn.commit()