nominatim/db/status.py

   1 """
   2 Access and helper functions for the status table.
   3 """
   4 import datetime as dt
   5 import logging
   6 import re
   7
   8 from ..tools.exec_utils import get_url
   9
  10 LOG = logging.getLogger()
  11
  12 def compute_database_date(conn):
  13     """ Determine the date of the database from the newest object in the
  14         data base.
  15     """
  16     # First, find the node with the highest ID in the database
  17     with conn.cursor() as cur:
  18         osmid = cur.scalar("SELECT max(osm_id) FROM place WHERE osm_type='N'")
  19
  20         if osmid is None:
  21             LOG.fatal("No data found in the database.")
  22             raise RuntimeError("No data found in the database.")
  23
  24     LOG.info("Using node id %d for timestamp lookup", osmid)
  25     # Get the node from the API to find the timestamp when it was created.
  26     node_url = 'https://www.openstreetmap.org/api/0.6/node/{}/1'.format(osmid)
  27     data = get_url(node_url)
  28
  29     match = re.search(r'timestamp="((\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2}))Z"', data)
  30
  31     if match is None:
  32         LOG.fatal("The node data downloaded from the API does not contain valid data.\n"
  33                   "URL used: %s", node_url)
  34         raise RuntimeError("Bad API data.")
  35
  36     LOG.debug("Found timestamp %s", match[1])
  37
  38     return dt.datetime.fromisoformat(match[1]).replace(tzinfo=dt.timezone.utc)
  39
  40
  41 def set_status(conn, date, seq=None, indexed=True):
  42     """ Replace the current status with the given status.
  43     """
  44     assert date.tzinfo == dt.timezone.utc
  45     with conn.cursor() as cur:
  46         cur.execute("TRUNCATE TABLE import_status")
  47         cur.execute("""INSERT INTO import_status (lastimportdate, sequence_id, indexed)
  48                        VALUES (%s, %s, %s)""", (date, seq, indexed))
  49
  50     conn.commit()