mirror of
https://github.com/osm-search/Nominatim.git
synced 2026-02-16 15:47:58 +00:00
65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
# SPDX-License-Identifier: GPL-2.0-only
|
|
#
|
|
# This file is part of Nominatim. (https://nominatim.org)
|
|
#
|
|
# Copyright (C) 2023 by the Nominatim developer community.
|
|
# For a full list of authors see the git log.
|
|
"""
|
|
Classes and function releated to status call.
|
|
"""
|
|
from typing import Optional, cast
|
|
import datetime as dt
|
|
import dataclasses
|
|
|
|
import sqlalchemy as sa
|
|
import asyncpg
|
|
|
|
from nominatim.api.connection import SearchConnection
|
|
from nominatim import version
|
|
|
|
@dataclasses.dataclass
|
|
class StatusResult:
|
|
""" Result of a call to the status API.
|
|
"""
|
|
status: int
|
|
message: str
|
|
software_version = version.NOMINATIM_VERSION
|
|
data_updated: Optional[dt.datetime] = None
|
|
database_version: Optional[version.NominatimVersion] = None
|
|
|
|
|
|
async def _get_database_date(conn: SearchConnection) -> Optional[dt.datetime]:
|
|
""" Query the database date.
|
|
"""
|
|
sql = sa.text('SELECT lastimportdate FROM import_status LIMIT 1')
|
|
result = await conn.execute(sql)
|
|
|
|
for row in result:
|
|
return cast(dt.datetime, row[0])
|
|
|
|
return None
|
|
|
|
|
|
async def _get_database_version(conn: SearchConnection) -> Optional[version.NominatimVersion]:
|
|
sql = sa.text("""SELECT value FROM nominatim_properties
|
|
WHERE property = 'database_version'""")
|
|
result = await conn.execute(sql)
|
|
|
|
for row in result:
|
|
return version.parse_version(cast(str, row[0]))
|
|
|
|
return None
|
|
|
|
|
|
async def get_status(conn: SearchConnection) -> StatusResult:
|
|
""" Execute a status API call.
|
|
"""
|
|
status = StatusResult(0, 'OK')
|
|
try:
|
|
status.data_updated = await _get_database_date(conn)
|
|
status.database_version = await _get_database_version(conn)
|
|
except asyncpg.PostgresError:
|
|
return StatusResult(700, 'Database connection failed')
|
|
|
|
return status
|