nominatim/server/falcon/server.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2023 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Server implementation using the falcon webserver framework.
   9 """
  10 from typing import Optional, Mapping, cast, Any
  11 from pathlib import Path
  12
  13 import falcon
  14 from falcon.asgi import App, Request, Response
  15
  16 from nominatim.api import NominatimAPIAsync
  17 import nominatim.api.v1 as api_impl
  18 from nominatim.config import Configuration
  19
  20 class HTTPNominatimError(Exception):
  21     """ A special exception class for errors raised during processing.
  22     """
  23     def __init__(self, msg: str, status: int, content_type: str) -> None:
  24         self.msg = msg
  25         self.status = status
  26         self.content_type = content_type
  27
  28
  29 async def nominatim_error_handler(req: Request, resp: Response,
  30                                   exception: HTTPNominatimError,
  31                                   _: Any) -> None:
  32     resp.status = exception.status
  33     resp.text = exception.msg
  34     resp.content_type = exception.content_type
  35
  36
  37 class ParamWrapper(api_impl.ASGIAdaptor):
  38     """ Adaptor class for server glue to Falcon framework.
  39     """
  40
  41     def __init__(self, req: Request, resp: Response,
  42                  config: Configuration) -> None:
  43         self.request = req
  44         self.response = resp
  45         self._config = config
  46
  47
  48     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  49         return cast(Optional[str], self.request.get_param(name, default=default))
  50
  51
  52     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  53         return cast(Optional[str], self.request.get_header(name, default=default))
  54
  55
  56     def error(self, msg: str, status: int = 400) -> HTTPNominatimError:
  57         return HTTPNominatimError(msg, status, self.content_type)
  58
  59
  60     def create_response(self, status: int, output: str) -> None:
  61         self.response.status = status
  62         self.response.text = output
  63         self.response.content_type = self.content_type
  64
  65
  66     def config(self) -> Configuration:
  67         return self._config
  68
  69
  70 class EndpointWrapper:
  71     """ Converter for server glue endpoint functions to Falcon request handlers.
  72     """
  73
  74     def __init__(self, func: api_impl.EndpointFunc, api: NominatimAPIAsync) -> None:
  75         self.func = func
  76         self.api = api
  77
  78
  79     async def on_get(self, req: Request, resp: Response) -> None:
  80         """ Implementation of the endpoint.
  81         """
  82         await self.func(self.api, ParamWrapper(req, resp, self.api.config))
  83
  84
  85 def get_application(project_dir: Path,
  86                     environ: Optional[Mapping[str, str]] = None) -> App:
  87     """ Create a Nominatim Falcon ASGI application.
  88     """
  89     api = NominatimAPIAsync(project_dir, environ)
  90
  91     app = App(cors_enable=api.config.get_bool('CORS_NOACCESSCONTROL'))
  92     app.add_error_handler(HTTPNominatimError, nominatim_error_handler)
  93
  94     legacy_urls = api.config.get_bool('SERVE_LEGACY_URLS')
  95     for name, func in api_impl.ROUTES:
  96         endpoint = EndpointWrapper(func, api)
  97         app.add_route(f"/{name}", endpoint)
  98         if legacy_urls:
  99             app.add_route(f"/{name}.php", endpoint)
 100
 101     return app