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         self.software_version = version.NOMINATIM_VERSION
  27         self.data_updated: Optional[dt.datetime]  = None
  28         self.database_version: Optional[version.NominatimVersion] = None
  29
  30
  31 async def _get_database_date(conn: AsyncConnection) -> Optional[dt.datetime]:
  32     """ Query the database date.
  33     """
  34     sql = sqla.text('SELECT lastimportdate FROM import_status LIMIT 1')
  35     result = await conn.execute(sql)
  36
  37     for row in result:
  38         return cast(dt.datetime, row[0])
  39
  40     return None
  41
  42
  43 async def _get_database_version(conn: AsyncConnection) -> Optional[version.NominatimVersion]:
  44     sql = sqla.text("""SELECT value FROM nominatim_properties
  45                        WHERE property = 'database_version'""")
  46     result = await conn.execute(sql)
  47
  48     for row in result:
  49         return version.parse_version(cast(str, row[0]))
  50
  51     return None
  52
  53
  54 async def get_status(engine: AsyncEngine) -> StatusResult:
  55     """ Execute a status API call.
  56     """
  57     status = StatusResult(0, 'OK')
  58     try:
  59         async with engine.begin() as conn:
  60             status.data_updated = await _get_database_date(conn)
  61             status.database_version = await _get_database_version(conn)
  62     except asyncpg.PostgresError:
  63         return StatusResult(700, 'Database connection failed')
  64
  65     return status