[mypy]
+plugins = sqlalchemy.ext.mypy.plugin
[mypy-icu.*]
ignore_missing_imports = True
-[mypy-osmium.*]
+[mypy-asyncpg.*]
ignore_missing_imports = True
[mypy-datrie.*]
--- /dev/null
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Implementation of classes for API access via libraries.
+"""
+from typing import Mapping, Optional, TypeVar, Callable, Any
+import functools
+import asyncio
+from pathlib import Path
+
+from sqlalchemy.engine.url import URL
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from nominatim.typing import StrPath
+from nominatim.config import Configuration
+from nominatim.apicmd.status import get_status, StatusResult
+
+class NominatimAPIAsync:
+ """ API loader asynchornous version.
+ """
+ def __init__(self, project_dir: Path,
+ environ: Optional[Mapping[str, str]] = None) -> None:
+ self.config = Configuration(project_dir, environ)
+
+ dsn = self.config.get_database_params()
+
+ dburl = URL.create(
+ 'postgresql+asyncpg',
+ database=dsn.get('dbname'),
+ username=dsn.get('user'), password=dsn.get('password'),
+ host=dsn.get('host'), port=int(dsn['port']) if 'port' in dsn else None,
+ query={k: v for k, v in dsn.items()
+ if k not in ('user', 'password', 'dbname', 'host', 'port')})
+ self.engine = create_async_engine(dburl,
+ connect_args={"server_settings": {"jit": "off"}},
+ future=True)
+
+
+ async def status(self) -> StatusResult:
+ """ Return the status of the database.
+ """
+ return await get_status(self.engine)
+
+
+class NominatimAPI:
+ """ API loader, synchronous version.
+ """
+
+ def __init__(self, project_dir: Path,
+ environ: Optional[Mapping[str, str]] = None) -> None:
+ self.async_api = NominatimAPIAsync(project_dir, environ)
+
+
+ def status(self) -> StatusResult:
+ return asyncio.get_event_loop().run_until_complete(self.async_api.status())
--- /dev/null
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 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 sqlalchemy as sqla
+from sqlalchemy.ext.asyncio.engine import AsyncEngine, AsyncConnection
+import asyncpg
+
+from nominatim import version
+
+class StatusResult:
+ """ Result of a call to the status API.
+ """
+
+ def __init__(self, status: int, msg: str):
+ self.status = status
+ self.message = msg
+ # XXX versions really should stay tuples here
+ self.software_version = version.version_str()
+ self.data_updated: Optional[dt.datetime] = None
+ self.database_version: Optional[str] = None
+
+
+async def _get_database_date(conn: AsyncConnection) -> Optional[dt.datetime]:
+ """ Query the database date.
+ """
+ sql = sqla.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: AsyncConnection) -> Optional[str]:
+ sql = sqla.text("""SELECT value FROM nominatim_properties
+ WHERE property = 'database_version'""")
+ result = await conn.execute(sql)
+
+ for row in result:
+ return cast(str, row[0])
+
+ return None
+
+
+async def get_status(engine: AsyncEngine) -> StatusResult:
+ """ Execute a status API call.
+ """
+ status = StatusResult(0, 'OK')
+ try:
+ async with engine.begin() as conn:
+ status.data_updated = await _get_database_date(conn)
+ status.database_version = await _get_database_version(conn)
+ except asyncpg.PostgresError as err:
+ return StatusResult(700, str(err))
+
+ return status
from nominatim.tools.exec_utils import run_api_script
from nominatim.errors import UsageError
from nominatim.clicmd.args import NominatimArgs
+from nominatim.api import NominatimAPI
+from nominatim.apicmd.status import StatusResult
+import nominatim.result_formatter.v1 as formatting
# Do not repeat documentation of subcommand classes.
# pylint: disable=C0111
class APIStatus:
- """\
+ """
Execute API status query.
This command works exactly the same as if calling the /status endpoint on
"""
def add_args(self, parser: argparse.ArgumentParser) -> None:
+ formats = formatting.create(StatusResult).list_formats()
group = parser.add_argument_group('API parameters')
- group.add_argument('--format', default='text', choices=['text', 'json'],
+ group.add_argument('--format', default=formats[0], choices=formats,
help='Format of result')
def run(self, args: NominatimArgs) -> int:
- return _run_api('status', args, dict(format=args.format))
+ status = NominatimAPI(args.project_dir).status()
+ print(formatting.create(StatusResult).format(status, args.format))
+ return 0
import yaml
from dotenv import dotenv_values
+from psycopg2.extensions import parse_dsn
from nominatim.typing import StrPath
from nominatim.errors import UsageError
Nominatim uses dotenv to configure the software. Configuration options
are resolved in the following order:
- * from the OS environment (or the dirctionary given in `environ`
+ * from the OS environment (or the dictionary given in `environ`)
* from the .env file in the project directory of the installation
* from the default installation in the configuration directory
return dsn
+ def get_database_params(self) -> Mapping[str, str]:
+ """ Get the configured parameters for the database connection
+ as a mapping.
+ """
+ dsn = self.DATABASE_DSN
+
+ if dsn.startswith('pgsql:'):
+ return dict((p.split('=', 1) for p in dsn[6:].split(';')))
+
+ return parse_dsn(dsn)
+
+
def get_import_style_file(self) -> Path:
""" Return the import style file as a path object. Translates the
name of the standard styles automatically into a file in the
--- /dev/null
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Helper classes and function for writing result formatting modules.
+"""
+from typing import Type, TypeVar, Dict, Mapping, List, Callable, Generic, Any
+from collections import defaultdict
+
+T = TypeVar('T') # pylint: disable=invalid-name
+FormatFunc = Callable[[T], str]
+
+class ResultFormatter(Generic[T]):
+ """ This class dispatches format calls to the appropriate formatting
+ function previously defined with the `format_func` decorator.
+ """
+
+ def __init__(self, funcs: Mapping[str, FormatFunc[T]]) -> None:
+ self.functions = funcs
+
+
+ def list_formats(self) -> List[str]:
+ """ Return a list of formats supported by this formatter.
+ """
+ return list(self.functions.keys())
+
+
+ def format(self, result: T, fmt: str) -> str:
+ """ Convert the given result into a string using the given format.
+
+ The format is expected to be in the list returned by
+ `list_formats()`.
+ """
+ return self.functions[fmt](result)
+
+
+class FormatDispatcher:
+ """ A factory class for result formatters.
+ """
+
+ def __init__(self) -> None:
+ self.format_functions: Dict[Type[Any], Dict[str, FormatFunc[Any]]] = defaultdict(dict)
+
+
+ def format_func(self, result_class: Type[T],
+ fmt: str) -> Callable[[FormatFunc[T]], FormatFunc[T]]:
+ """ Decorator for a function that formats a given type of result into the
+ selected format.
+ """
+ def decorator(func: FormatFunc[T]) -> FormatFunc[T]:
+ self.format_functions[result_class][fmt] = func
+ return func
+
+ return decorator
+
+
+ def __call__(self, result_class: Type[T]) -> ResultFormatter[T]:
+ """ Create an instance of a format class for the given result type.
+ """
+ return ResultFormatter(self.format_functions[result_class])
--- /dev/null
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Output formatters for API version v1.
+"""
+from typing import Dict, Any
+from collections import OrderedDict
+import json
+
+from nominatim.result_formatter.base import FormatDispatcher
+from nominatim.apicmd.status import StatusResult
+
+create = FormatDispatcher()
+
+@create.format_func(StatusResult, 'text')
+def _format_status_text(result: StatusResult) -> str:
+ return result.message
+
+
+@create.format_func(StatusResult, 'json')
+def _format_status_json(result: StatusResult) -> str:
+ # XXX write a simple JSON serializer
+ out: Dict[str, Any] = OrderedDict()
+ out['status'] = result.status
+ out['message'] = result.message
+ if result.data_updated is not None:
+ out['data_updated'] = result.data_updated
+ out['software_version'] = result.software_version
+ if result.database_version is not None:
+ out['database_version'] = result.database_version
+
+ return json.dumps(out)
projects / nominatim.git / commitdiff