src/nominatim_db/db/status.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2024 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Access and helper functions for the status and status log table.
   9 """
  10 from typing import Optional, Tuple
  11 import datetime as dt
  12 import logging
  13 import re
  14
  15 from .connection import Connection, table_exists, execute_scalar
  16 from ..utils.url_utils import get_url
  17 from ..errors import UsageError
  18
  19 LOG = logging.getLogger()
  20 ISODATE_FORMAT = '%Y-%m-%dT%H:%M:%S'
  21
  22
  23 def compute_database_date(conn: Connection, offline: bool = False) -> dt.datetime:
  24     """ Determine the date of the database from the newest object in the
  25         data base.
  26     """
  27     # If there is a date from osm2pgsql available, use that.
  28     if table_exists(conn, 'osm2pgsql_properties'):
  29         with conn.cursor() as cur:
  30             cur.execute(""" SELECT value FROM osm2pgsql_properties
  31                             WHERE property = 'current_timestamp' """)
  32             row = cur.fetchone()
  33             if row is not None:
  34                 return dt.datetime.strptime(row[0], "%Y-%m-%dT%H:%M:%SZ")\
  35                                   .replace(tzinfo=dt.timezone.utc)
  36
  37     if offline:
  38         raise UsageError("Cannot determine database date from data in offline mode.")
  39
  40     # Else, find the node with the highest ID in the database
  41     if table_exists(conn, 'place'):
  42         osmid = execute_scalar(conn, "SELECT max(osm_id) FROM place WHERE osm_type='N'")
  43     else:
  44         osmid = execute_scalar(conn, "SELECT max(osm_id) FROM placex WHERE osm_type='N'")
  45
  46     if osmid is None:
  47         LOG.fatal("No data found in the database.")
  48         raise UsageError("No data found in the database.")
  49
  50     LOG.info("Using node id %d for timestamp lookup", osmid)
  51     # Get the node from the API to find the timestamp when it was created.
  52     node_url = f'https://www.openstreetmap.org/api/0.6/node/{osmid}/1'
  53     data = get_url(node_url)
  54
  55     match = re.search(r'timestamp="((\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2}))Z"', data)
  56
  57     if match is None:
  58         LOG.fatal("The node data downloaded from the API does not contain valid data.\n"
  59                   "URL used: %s", node_url)
  60         raise UsageError("Bad API data.")
  61
  62     LOG.debug("Found timestamp %s", match.group(1))
  63
  64     return dt.datetime.strptime(match.group(1), ISODATE_FORMAT).replace(tzinfo=dt.timezone.utc)
  65
  66
  67 def set_status(conn: Connection, date: Optional[dt.datetime],
  68                seq: Optional[int] = None, indexed: bool = True) -> None:
  69     """ Replace the current status with the given status. If date is `None`
  70         then only sequence and indexed will be updated as given. Otherwise
  71         the whole status is replaced.
  72         The change will be committed to the database.
  73     """
  74     assert date is None or date.tzinfo == dt.timezone.utc
  75     with conn.cursor() as cur:
  76         if date is None:
  77             cur.execute("UPDATE import_status set sequence_id = %s, indexed = %s",
  78                         (seq, indexed))
  79         else:
  80             cur.execute("TRUNCATE TABLE import_status")
  81             cur.execute("""INSERT INTO import_status (lastimportdate, sequence_id, indexed)
  82                            VALUES (%s, %s, %s)""", (date, seq, indexed))
  83
  84     conn.commit()
  85
  86
  87 def get_status(conn: Connection) -> Tuple[Optional[dt.datetime], Optional[int], Optional[bool]]:
  88     """ Return the current status as a triple of (date, sequence, indexed).
  89         If status has not been set up yet, a triple of None is returned.
  90     """
  91     with conn.cursor() as cur:
  92         cur.execute("SELECT * FROM import_status LIMIT 1")
  93         if cur.rowcount < 1:
  94             return None, None, None
  95
  96         row = cur.fetchone()
  97         assert row
  98         return row.lastimportdate, row.sequence_id, row.indexed
  99
 100
 101 def set_indexed(conn: Connection, state: bool) -> None:
 102     """ Set the indexed flag in the status table to the given state.
 103     """
 104     with conn.cursor() as cur:
 105         cur.execute("UPDATE import_status SET indexed = %s", (state, ))
 106     conn.commit()
 107
 108
 109 def log_status(conn: Connection, start: dt.datetime,
 110                event: str, batchsize: Optional[int] = None) -> None:
 111     """ Write a new status line to the `import_osmosis_log` table.
 112     """
 113     with conn.cursor() as cur:
 114         cur.execute("""INSERT INTO import_osmosis_log
 115                        (batchend, batchseq, batchsize, starttime, endtime, event)
 116                        SELECT lastimportdate, sequence_id, %s, %s, now(), %s FROM import_status""",
 117                     (batchsize, start, event))
 118     conn.commit()