X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/6e89310a9285f1ad15d8002bf68f578eada367a0..122ecd46269d48a8b2aacba2474311c0400d2a9d:/src/nominatim_api/result_formatting.py diff --git a/src/nominatim_api/result_formatting.py b/src/nominatim_api/result_formatting.py index fc22fc0f..0f91b761 100644 --- a/src/nominatim_api/result_formatting.py +++ b/src/nominatim_api/result_formatting.py @@ -7,22 +7,30 @@ """ Helper classes and functions for formatting results into API responses. """ -from typing import Type, TypeVar, Dict, List, Callable, Any, Mapping +from typing import Type, TypeVar, Dict, List, Callable, Any, Mapping, Optional, cast from collections import defaultdict +from pathlib import Path +import importlib -T = TypeVar('T') # pylint: disable=invalid-name +from .server.content_types import CONTENT_JSON + +T = TypeVar('T') FormatFunc = Callable[[T, Mapping[str, Any]], str] +ErrorFormatFunc = Callable[[str, str, int], str] class FormatDispatcher: - """ Helper class to conveniently create formatting functions in - a module using decorators. + """ Container for formatting functions for results. + Functions can conveniently be added by using decorated functions. """ - 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) - 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 @@ -34,19 +42,24 @@ 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. """ 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, options: Mapping[str, Any]) -> str: """ Convert the given result into a string using the given format. @@ -54,3 +67,53 @@ class FormatDispatcher: `list_formats()`. """ 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. + + The function first tries to find a module api//format.py + in the project directory. This file must export a single variable + `dispatcher`. + + If the function does not exist, the default formatter is loaded. + """ + if project_dir is not None: + priv_module = project_dir / 'api' / api_name / 'format.py' + if priv_module.is_file(): + spec = importlib.util.spec_from_file_location(f'api.{api_name},format', + str(priv_module)) + if spec: + module = importlib.util.module_from_spec(spec) + # Do not add to global modules because there is no standard + # module name that Python can resolve. + assert spec.loader is not None + spec.loader.exec_module(module) + + return cast(FormatDispatcher, module.dispatch) + + return cast(FormatDispatcher, + importlib.import_module(f'nominatim_api.{api_name}.format').dispatch)