#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2022 by the Nominatim developer community.
+# Copyright (C) 2023 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
-Helper classes and function for writing result formatting modules.
+Helper classes and functions for formating results into API responses.
"""
-from typing import Type, TypeVar, Dict, Mapping, List, Callable, Generic, Any
+from typing import Type, TypeVar, Dict, List, Callable, 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 supports_format(self, fmt: str) -> bool:
- """ Check if the given format is supported by this formatter.
- """
- return fmt in self.functions
-
-
- 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.
+ """ Helper class to conveniently create formatting functions in
+ a module using decorators.
"""
def __init__(self) -> None:
return decorator
- def __call__(self, result_class: Type[T]) -> ResultFormatter[T]:
- """ Create an instance of a format class for the given result type.
+ def list_formats(self, result_type: Type[Any]) -> List[str]:
+ """ Return a list of formats supported by this formatter.
+ """
+ return list(self.format_functions[result_type].keys())
+
+
+ def supports_format(self, result_type: Type[Any], fmt: str) -> bool:
+ """ Check if the given format is supported by this formatter.
+ """
+ return fmt in self.format_functions[result_type]
+
+
+ def format_result(self, result: Any, 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 ResultFormatter(self.format_functions[result_class])
+ return self.format_functions[type(result)][fmt](result)
--- /dev/null
+# 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.
+"""
+Implementation of API version v1 (aka the legacy version).
+"""
+
+import nominatim.api.v1.format as _format
+
+list_formats = _format.dispatch.list_formats
+supports_format = _format.dispatch.supports_format
+format_result = _format.dispatch.format_result
from collections import OrderedDict
import json
-from nominatim.result_formatter.base import FormatDispatcher
+from nominatim.api.result_formatting import FormatDispatcher
from nominatim.api import StatusResult
-create = FormatDispatcher()
+dispatch = FormatDispatcher()
-@create.format_func(StatusResult, 'text')
+@dispatch.format_func(StatusResult, 'text')
def _format_status_text(result: StatusResult) -> str:
if result.status:
return f"ERROR: {result.message}"
return 'OK'
-@create.format_func(StatusResult, 'json')
+@dispatch.format_func(StatusResult, 'json')
def _format_status_json(result: StatusResult) -> str:
out: Dict[str, Any] = OrderedDict()
out['status'] = result.status
from nominatim.errors import UsageError
from nominatim.clicmd.args import NominatimArgs
from nominatim.api import NominatimAPI, StatusResult
-import nominatim.result_formatter.v1 as formatting
+import nominatim.api.v1 as api_output
# Do not repeat documentation of subcommand classes.
# pylint: disable=C0111
"""
def add_args(self, parser: argparse.ArgumentParser) -> None:
- formats = formatting.create(StatusResult).list_formats()
+ formats = api_output.list_formats(StatusResult)
group = parser.add_argument_group('API parameters')
group.add_argument('--format', default=formats[0], choices=formats,
help='Format of result')
def run(self, args: NominatimArgs) -> int:
status = NominatimAPI(args.project_dir).status()
- print(formatting.create(StatusResult).format(status, args.format))
+ print(api_output.format_result(status, args.format))
return 0
import falcon.asgi
from nominatim.api import NominatimAPIAsync, StatusResult
-import nominatim.result_formatter.v1 as formatting
+import nominatim.api.v1 as api_impl
CONTENT_TYPE = {
'text': falcon.MEDIA_TEXT,
def __init__(self, project_dir: Path, environ: Optional[Mapping[str, str]]) -> None:
self.api = NominatimAPIAsync(project_dir, environ)
- self.formatters = {}
-
- for rtype in (StatusResult, ):
- self.formatters[rtype] = formatting.create(rtype)
def parse_format(self, req: falcon.asgi.Request, rtype: Type[Any], default: str) -> None:
format value to assume when no parameter is present.
"""
req.context.format = req.get_param('format', default=default)
- req.context.formatter = self.formatters[rtype]
- if not req.context.formatter.supports_format(req.context.format):
+ if not api_impl.supports_format(rtype, req.context.format):
raise falcon.HTTPBadRequest(
description="Parameter 'format' must be one of: " +
- ', '.join(req.context.formatter.list_formats()))
+ ', '.join(api_impl.list_formats(rtype)))
def format_response(self, req: falcon.asgi.Request, resp: falcon.asgi.Response,
""" Render response into a string according to the formatter
set in `parse_format()`.
"""
- resp.text = req.context.formatter.format(result, req.context.format)
+ resp.text = api_impl.format_result(result, req.context.format)
resp.content_type = CONTENT_TYPE.get(req.context.format, falcon.MEDIA_JSON)
import sanic
from nominatim.api import NominatimAPIAsync, StatusResult
-import nominatim.result_formatter.v1 as formatting
+import nominatim.api.v1 as api_impl
api = sanic.Blueprint('NominatimAPI')
""" Render a response from the query results using the configured
formatter.
"""
- body = request.ctx.formatter.format(result, request.ctx.format)
+ body = api_impl.format_result(result, request.ctx.format)
return sanic.response.text(body,
content_type=CONTENT_TYPE.get(request.ctx.format,
'application/json'))
is present.
"""
assert request.route is not None
- request.ctx.formatter = request.app.ctx.formatters[request.route.ctx.result_type]
request.ctx.format = request.args.get('format', request.route.ctx.default_format)
- if not request.ctx.formatter.supports_format(request.ctx.format):
+ if not api_impl.supports_format(request.route.ctx.result_type, request.ctx.format):
return usage_error("Parameter 'format' must be one of: " +
- ', '.join(request.ctx.formatter.list_formats()))
+ ', '.join(api_impl.list_formats(request.route.ctx.result_type)))
return None
app = sanic.Sanic("NominatimInstance")
app.ctx.api = NominatimAPIAsync(project_dir, environ)
- app.ctx.formatters = {}
- for rtype in (StatusResult, ):
- app.ctx.formatters[rtype] = formatting.create(rtype)
app.blueprint(api)
from starlette.requests import Request
from nominatim.api import NominatimAPIAsync, StatusResult
-import nominatim.result_formatter.v1 as formatting
+import nominatim.api.v1 as api_impl
CONTENT_TYPE = {
'text': 'text/plain; charset=utf-8',
'xml': 'text/xml; charset=utf-8'
}
-FORMATTERS = {
- StatusResult: formatting.create(StatusResult)
-}
-
-
def parse_format(request: Request, rtype: Type[Any], default: str) -> None:
""" Get and check the 'format' parameter and prepare the formatter.
`rtype` describes the expected return type and `default` the
format value to assume when no parameter is present.
"""
fmt = request.query_params.get('format', default=default)
- fmtter = FORMATTERS[rtype]
- if not fmtter.supports_format(fmt):
+ if not api_impl.supports_format(rtype, fmt):
raise HTTPException(400, detail="Parameter 'format' must be one of: " +
- ', '.join(fmtter.list_formats()))
+ ', '.join(api_impl.list_formats(rtype)))
request.state.format = fmt
- request.state.formatter = fmtter
def format_response(request: Request, result: Any) -> Response:
- """ Render response into a string according to the formatter
- set in `parse_format()`.
+ """ Render response into a string according.
"""
fmt = request.state.format
- return Response(request.state.formatter.format(result, fmt),
+ return Response(api_impl.format_result(result, fmt),
media_type=CONTENT_TYPE.get(fmt, 'application/json'))
--- /dev/null
+# 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.
+"""
+Tests for formatting results for the V1 API.
+"""
+import datetime as dt
+import pytest
+
+import nominatim.api.v1 as api_impl
+from nominatim.api import StatusResult
+from nominatim.version import NOMINATIM_VERSION
+
+STATUS_FORMATS = {'text', 'json'}
+
+# StatusResult
+
+def test_status_format_list():
+ assert set(api_impl.list_formats(StatusResult)) == STATUS_FORMATS
+
+
+@pytest.mark.parametrize('fmt', list(STATUS_FORMATS))
+def test_status_supported(fmt):
+ assert api_impl.supports_format(StatusResult, fmt)
+
+
+def test_status_unsupported():
+ assert not api_impl.supports_format(StatusResult, 'gagaga')
+
+
+def test_status_format_text():
+ assert api_impl.format_result(StatusResult(0, 'message here'), 'text') == 'OK'
+
+
+def test_status_format_text():
+ assert api_impl.format_result(StatusResult(500, 'message here'), 'text') == 'ERROR: message here'
+
+
+def test_status_format_json_minimal():
+ status = StatusResult(700, 'Bad format.')
+
+ result = api_impl.format_result(status, 'json')
+
+ assert result == '{"status": 700, "message": "Bad format.", "software_version": "%s"}' % (NOMINATIM_VERSION, )
+
+
+def test_status_format_json_full():
+ status = StatusResult(0, 'OK')
+ status.data_updated = dt.datetime(2010, 2, 7, 20, 20, 3, 0, tzinfo=dt.timezone.utc)
+ status.database_version = '5.6'
+
+ result = api_impl.format_result(status, 'json')
+
+ assert result == '{"status": 0, "message": "OK", "data_updated": "2010-02-07T20:20:03+00:00", "software_version": "%s", "database_version": "5.6"}' % (NOMINATIM_VERSION, )
+++ /dev/null
-# 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.
-"""
-Tests for formatting results for the V1 API.
-"""
-import datetime as dt
-import pytest
-
-import nominatim.result_formatter.v1 as format_module
-from nominatim.api import StatusResult
-from nominatim.version import NOMINATIM_VERSION
-
-STATUS_FORMATS = {'text', 'json'}
-
-class TestStatusResultFormat:
-
-
- @pytest.fixture(autouse=True)
- def make_formatter(self):
- self.formatter = format_module.create(StatusResult)
-
-
- def test_format_list(self):
- assert set(self.formatter.list_formats()) == STATUS_FORMATS
-
-
- @pytest.mark.parametrize('fmt', list(STATUS_FORMATS))
- def test_supported(self, fmt):
- assert self.formatter.supports_format(fmt)
-
-
- def test_unsupported(self):
- assert not self.formatter.supports_format('gagaga')
-
-
- def test_format_text(self):
- assert self.formatter.format(StatusResult(0, 'message here'), 'text') == 'OK'
-
-
- def test_format_text(self):
- assert self.formatter.format(StatusResult(500, 'message here'), 'text') == 'ERROR: message here'
-
-
- def test_format_json_minimal(self):
- status = StatusResult(700, 'Bad format.')
-
- result = self.formatter.format(status, 'json')
-
- assert result == '{"status": 700, "message": "Bad format.", "software_version": "%s"}' % (NOMINATIM_VERSION, )
-
-
- def test_format_json_full(self):
- status = StatusResult(0, 'OK')
- status.data_updated = dt.datetime(2010, 2, 7, 20, 20, 3, 0, tzinfo=dt.timezone.utc)
- status.database_version = '5.6'
-
- result = self.formatter.format(status, 'json')
-
- assert result == '{"status": 0, "message": "OK", "data_updated": "2010-02-07T20:20:03+00:00", "software_version": "%s", "database_version": "5.6"}' % (NOMINATIM_VERSION, )