configurable error formatting and content type in result formatter

src/nominatim_api/result_formatting.py

@@ -12,8 +12,11 @@ from collections import defaultdict
 from pathlib import Path
 import importlib
 
+from .server.content_types import CONTENT_JSON
+
 T = TypeVar('T') # pylint: disable=invalid-name
 FormatFunc = Callable[[T, Mapping[str, Any]], str]
+ErrorFormatFunc = Callable[[str, str, int], str]
 
 
 class FormatDispatcher:
@@ -21,7 +24,11 @@ class FormatDispatcher:
         a module using decorators.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, content_types: Optional[Mapping[str, str]] = None) -> None:
+        self.error_handler: ErrorFormatFunc = lambda ct, msg, status: f"ERROR {status}: {msg}"
+        self.content_types: Dict[str, str] = {}
+        if content_types:
+            self.content_types.update(content_types)
         self.format_functions: Dict[Type[Any], Dict[str, FormatFunc[Any]]] = defaultdict(dict)
 
 
@@ -37,6 +44,15 @@ class FormatDispatcher:
         return decorator
 
 
+    def error_format_func(self, func: ErrorFormatFunc) -> ErrorFormatFunc:
+        """ Decorator for a function that formats error messges.
+            There is only one error formatter per dispatcher. Using
+            the decorator repeatedly will overwrite previous functions.
+        """
+        self.error_handler = func
+        return func
+
+
     def list_formats(self, result_type: Type[Any]) -> List[str]:
         """ Return a list of formats supported by this formatter.
         """
@@ -58,6 +74,32 @@ class FormatDispatcher:
         return self.format_functions[type(result)][fmt](result, options)
 
 
+    def format_error(self, content_type: str, msg: str, status: int) -> str:
+        """ Convert the given error message into a response string
+            taking the requested content_type into account.
+
+            Change the format using the error_format_func decorator.
+        """
+        return self.error_handler(content_type, msg, status)
+
+
+    def set_content_type(self, fmt: str, content_type: str) -> None:
+        """ Set the content type for the given format. This is the string
+            that will be returned in the Content-Type header of the HTML
+            response, when the given format is choosen.
+        """
+        self.content_types[fmt] = content_type
+
+
+    def get_content_type(self, fmt: str) -> str:
+        """ Return the content type for the given format.
+
+            If no explicit content type has been defined, then
+            JSON format is assumed.
+        """
+        return self.content_types.get(fmt, CONTENT_JSON)
+
+
 def load_format_dispatcher(api_name: str, project_dir: Optional[Path]) -> FormatDispatcher:
     """ Load the dispatcher for the given API.
 

src/nominatim_api/server/asgi_adaptor.py

@@ -12,16 +12,9 @@ import abc
 import math
 
 from ..config import Configuration
-from .. import logging as loglib
 from ..core import NominatimAPIAsync
 from ..result_formatting import FormatDispatcher
-
-CONTENT_TEXT = 'text/plain; charset=utf-8'
-CONTENT_XML = 'text/xml; charset=utf-8'
-CONTENT_HTML = 'text/html; charset=utf-8'
-CONTENT_JSON = 'application/json; charset=utf-8'
-
-CONTENT_TYPE = {'text': CONTENT_TEXT, 'xml': CONTENT_XML, 'debug': CONTENT_HTML}
+from .content_types import CONTENT_TEXT
 
 class ASGIAdaptor(abc.ABC):
     """ Adapter class for the different ASGI frameworks.
@@ -156,22 +149,8 @@ class ASGIAdaptor(abc.ABC):
             message. The message will be formatted according to the
             output format chosen by the request.
         """
-        if self.content_type == CONTENT_XML:
-            msg = f"""<?xml version="1.0" encoding="UTF-8" ?>
-                      <error>
-                        <code>{status}</code>
-                        <message>{msg}</message>
-                      </error>
-                   """
-        elif self.content_type == CONTENT_JSON:
-            msg = f"""{{"error":{{"code":{status},"message":"{msg}"}}}}"""
-        elif self.content_type == CONTENT_HTML:
-            loglib.log().section('Execution error')
-            loglib.log().var_dump('Status', status)
-            loglib.log().var_dump('Message', msg)
-            msg = loglib.get_and_disable()
-
-        raise self.error(msg, status)
+        raise self.error(self.formatting().format_error(self.content_type, msg, status),
+                         status)
 
 
 EndpointFunc = Callable[[NominatimAPIAsync, ASGIAdaptor], Any]

src/nominatim_api/server/content_types.py

@@ -0,0 +1,14 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2024 by the Nominatim developer community.
+# For a full list of authors see the git log.
+"""
+Constants for various content types for server responses.
+"""
+
+CONTENT_TEXT = 'text/plain; charset=utf-8'
+CONTENT_XML = 'text/xml; charset=utf-8'
+CONTENT_HTML = 'text/html; charset=utf-8'
+CONTENT_JSON = 'application/json; charset=utf-8'

src/nominatim_api/v1/format.py

@@ -19,12 +19,38 @@ from ..localization import Locales
 from ..result_formatting import FormatDispatcher
 from .classtypes import ICONS
 from . import format_json, format_xml
+from .. import logging as loglib
+from ..server import content_types as ct
 
 class RawDataList(List[Dict[str, Any]]):
     """ Data type for formatting raw data lists 'as is' in json.
     """
 
-dispatch = FormatDispatcher()
+dispatch = FormatDispatcher({'text': ct.CONTENT_TEXT,
+                             'xml': ct.CONTENT_XML,
+                             'debug': ct.CONTENT_HTML})
+
+@dispatch.error_format_func
+def _format_error(content_type: str, msg: str, status: int) -> str:
+    if content_type == ct.CONTENT_XML:
+        return f"""<?xml version="1.0" encoding="UTF-8" ?>
+                   <error>
+                     <code>{status}</code>
+                     <message>{msg}</message>
+                   </error>
+                """
+
+    if content_type == ct.CONTENT_JSON:
+        return f"""{{"error":{{"code":{status},"message":"{msg}"}}}}"""
+
+    if content_type == ct.CONTENT_HTML:
+        loglib.log().section('Execution error')
+        loglib.log().var_dump('Status', status)
+        loglib.log().var_dump('Message', msg)
+        return loglib.get_and_disable()
+
+    return f"ERROR {status}: {msg}"
+
 
 @dispatch.format_func(StatusResult, 'text')
 def _format_status_text(result: StatusResult, _: Mapping[str, Any]) -> str:

src/nominatim_api/v1/server_glue.py

@@ -24,14 +24,15 @@ from ..status import StatusResult
 from ..results import DetailedResult, ReverseResults, SearchResult, SearchResults
 from ..localization import Locales
 from . import helpers
-from ..server.asgi_adaptor import CONTENT_HTML, CONTENT_JSON, CONTENT_TYPE, ASGIAdaptor
+from ..server import content_types as ct
+from ..server.asgi_adaptor import ASGIAdaptor
 
 def build_response(adaptor: ASGIAdaptor, output: str, status: int = 200,
                    num_results: int = 0) -> Any:
     """ Create a response from the given output. Wraps a JSONP function
         around the response, if necessary.
     """
-    if adaptor.content_type == CONTENT_JSON and status == 200:
+    if adaptor.content_type == ct.CONTENT_JSON and status == 200:
         jsonp = adaptor.get('json_callback')
         if jsonp is not None:
             if any(not part.isidentifier() for part in jsonp.split('.')):
@@ -57,7 +58,7 @@ def setup_debugging(adaptor: ASGIAdaptor) -> bool:
     """
     if adaptor.get_bool('debug', False):
         loglib.set_log_output('html')
-        adaptor.content_type = CONTENT_HTML
+        adaptor.content_type = ct.CONTENT_HTML
         return True
 
     return False
@@ -83,11 +84,13 @@ def parse_format(adaptor: ASGIAdaptor, result_type: Type[Any], default: str) ->
     fmt = adaptor.get('format', default=default)
     assert fmt is not None
 
-    if not adaptor.formatting().supports_format(result_type, fmt):
-        adaptor.raise_error("Parameter 'format' must be one of: " +
-                          ', '.join(adaptor.formatting().list_formats(result_type)))
+    formatting = adaptor.formatting()
 
-    adaptor.content_type = CONTENT_TYPE.get(fmt, CONTENT_JSON)
+    if not formatting.supports_format(result_type, fmt):
+        adaptor.raise_error("Parameter 'format' must be one of: " +
+                          ', '.join(formatting.list_formats(result_type)))
+
+    adaptor.content_type = formatting.get_content_type(fmt)
     return fmt
 
 

test/python/api/test_server_glue_v1.py

@@ -127,7 +127,7 @@ class TestAdaptorRaiseError:
         err = self.run_raise_error('TEST', 404)
 
         assert self.adaptor.content_type == 'text/plain; charset=utf-8'
-        assert err.msg == 'TEST'
+        assert err.msg == 'ERROR 404: TEST'
         assert err.status == 404