nominatim/apicmd/status.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Classes and function releated to status call.
   9 """
  10 from typing import Optional, cast
  11 import datetime as dt
  12
  13 import sqlalchemy as sqla
  14 from sqlalchemy.ext.asyncio.engine import AsyncEngine, AsyncConnection
  15 import asyncpg
  16
  17 from nominatim import version
  18
  19 class StatusResult:
  20     """ Result of a call to the status API.
  21     """
  22
  23     def __init__(self, status: int, msg: str):
  24         self.status = status
  25         self.message = msg
  26         # XXX versions really should stay tuples here
  27         self.software_version = version.version_str()
  28         self.data_updated: Optional[dt.datetime]  = None
  29         self.database_version: Optional[str] = None
  30
  31
  32 async def _get_database_date(conn: AsyncConnection) -> Optional[dt.datetime]:
  33     """ Query the database date.
  34     """
  35     sql = sqla.text('SELECT lastimportdate FROM import_status LIMIT 1')
  36     result = await conn.execute(sql)
  37
  38     for row in result:
  39         return cast(dt.datetime, row[0])
  40
  41     return None
  42
  43
  44 async def _get_database_version(conn: AsyncConnection) -> Optional[str]:
  45     sql = sqla.text("""SELECT value FROM nominatim_properties
  46                        WHERE property = 'database_version'""")
  47     result = await conn.execute(sql)
  48
  49     for row in result:
  50         return cast(str, row[0])
  51
  52     return None
  53
  54
  55 async def get_status(engine: AsyncEngine) -> StatusResult:
  56     """ Execute a status API call.
  57     """
  58     status = StatusResult(0, 'OK')
  59     try:
  60         async with engine.begin() as conn:
  61             status.data_updated = await _get_database_date(conn)
  62             status.database_version = await _get_database_version(conn)
  63     except asyncpg.PostgresError:
  64         return StatusResult(700, 'Database connection failed')
  65
  66     return status