Merge pull request #3356 from lonvia/use-date-from-osm2pgsql-prop

Use import date from osm2pgsql property table if available

Author: lonvia

nominatim/clicmd/setup.py

@@ -219,12 +219,11 @@ def _finalize_database(self, dsn: str, offline: bool) -> None:
         """ Determine the database date and set the status accordingly.
         """
         with connect(dsn) as conn:
-            if not offline:
-                try:
-                    dbdate = status.compute_database_date(conn)
-                    status.set_status(conn, dbdate)
-                    LOG.info('Database is at %s.', dbdate)
-                except Exception as exc: # pylint: disable=broad-except
-                    LOG.error('Cannot determine date of database: %s', exc)
-
             properties.set_property(conn, 'database_version', str(NOMINATIM_VERSION))
+
+            try:
+                dbdate = status.compute_database_date(conn, offline)
+                status.set_status(conn, dbdate)
+                LOG.info('Database is at %s.', dbdate)
+            except Exception as exc: # pylint: disable=broad-except
+                LOG.error('Cannot determine date of database: %s', exc)

nominatim/db/status.py

@@ -29,11 +29,24 @@ class StatusRow(TypedDict):
     indexed: Optional[bool]
 
 
-def compute_database_date(conn: Connection) -> dt.datetime:
+def compute_database_date(conn: Connection, offline: bool = False) -> 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
+    # If there is a date from osm2pgsql available, use that.
+    if conn.table_exists('osm2pgsql_properties'):
+        with conn.cursor() as cur:
+            cur.execute(""" SELECT value FROM osm2pgsql_properties
+                            WHERE property = 'current_timestamp' """)
+            row = cur.fetchone()
+            if row is not None:
+                return dt.datetime.strptime(row[0], "%Y-%m-%dT%H:%M:%SZ")\
+                                  .replace(tzinfo=dt.timezone.utc)
+
+    if offline:
+        raise UsageError("Cannot determine database date from data in offline mode.")
+
+    # Else, find the node with the highest ID in the database
     with conn.cursor() as cur:
         if conn.table_exists('place'):
             osmid = cur.scalar("SELECT max(osm_id) FROM place WHERE osm_type='N'")

test/python/db/test_status.py

@@ -31,6 +31,22 @@ def setup_status_table(status_table):
     pass
 
 
+@pytest.mark.parametrize('offline', [True, False])
+def test_compute_database_date_from_osm2pgsql(table_factory, temp_db_conn, offline):
+    table_factory('osm2pgsql_properties', 'property TEXT, value TEXT',
+                  content=(('current_timestamp', '2024-01-03T23:45:54Z'), ))
+
+    date = nominatim.db.status.compute_database_date(temp_db_conn, offline=offline)
+    assert date == iso_date('2024-01-03T23:45:54')
+
+
+def test_compute_database_date_from_osm2pgsql_nodata(table_factory, temp_db_conn):
+    table_factory('osm2pgsql_properties', 'property TEXT, value TEXT')
+
+    with pytest.raises(UsageError, match='Cannot determine database date from data in offline mode'):
+        nominatim.db.status.compute_database_date(temp_db_conn, offline=True)
+
+
 def test_compute_database_date_place_empty(place_table, temp_db_conn):
     with pytest.raises(UsageError):
         nominatim.db.status.compute_database_date(temp_db_conn)