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:
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)
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.
"""
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.
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.
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]
--- /dev/null
+# 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'
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:
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('.')):
"""
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
fmt = adaptor.get('format', default=default)
assert fmt is not None
- if not adaptor.formatting().supports_format(result_type, fmt):
+ formatting = adaptor.formatting()
+
+ if not formatting.supports_format(result_type, fmt):
adaptor.raise_error("Parameter 'format' must be one of: " +
- ', '.join(adaptor.formatting().list_formats(result_type)))
+ ', '.join(formatting.list_formats(result_type)))
- adaptor.content_type = CONTENT_TYPE.get(fmt, CONTENT_JSON)
+ adaptor.content_type = formatting.get_content_type(fmt)
return fmt
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
projects / nominatim.git / commitdiff