actions: add test for database migration

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

diff --git a/src/nominatim_db/db/status.py b/src/nominatim_db/db/status.py
index 1d2b3bec5e6bdebeab68d7c1d266cdf06f9257e1..4fe9f4449c1ce24e6af45feb34e8fad1bde9d9f8 100644 (file)
--- a/src/nominatim_db/db/status.py
+++ b/src/nominatim_db/db/status.py
@@ -7,7 +7,7 @@
 """
 Access and helper functions for the status and status log table.
 """
 """
 Access and helper functions for the status and status log table.
 """

-from typing import Optional, Tuple, cast
+from typing import Optional, Tuple

 import datetime as dt
 import logging
 import re
 import datetime as dt
 import logging
 import re


@@ -15,20 +15,11 @@ import re
 from .connection import Connection, table_exists, execute_scalar
 from ..utils.url_utils import get_url
 from ..errors import UsageError
 from .connection import Connection, table_exists, execute_scalar
 from ..utils.url_utils import get_url
 from ..errors import UsageError

-from ..typing import TypedDict

 
 LOG = logging.getLogger()
 ISODATE_FORMAT = '%Y-%m-%dT%H:%M:%S'
 
 
 
 LOG = logging.getLogger()
 ISODATE_FORMAT = '%Y-%m-%dT%H:%M:%S'
 
 

-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, offline: bool = False) -> dt.datetime:
     """ Determine the date of the database from the newest object in the
         data base.
 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.


@@ -102,8 +93,9 @@ def get_status(conn: Connection) -> Tuple[Optional[dt.datetime], Optional[int],
         if cur.rowcount < 1:
             return None, None, None
 
         if cur.rowcount < 1:
             return None, None, None
 

-        row = cast(StatusRow, cur.fetchone())
-        return row['lastimportdate'], row['sequence_id'], row['indexed']
+        row = cur.fetchone()
+        assert row
+        return row.lastimportdate, row.sequence_id, row.indexed

 
 
 def set_indexed(conn: Connection, state: bool) -> None:
 
 
 def set_indexed(conn: Connection, state: bool) -> None: