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 from falcon.asgi import App, Request, Response
  14
  15 from nominatim.api import NominatimAPIAsync
  16 import nominatim.api.v1 as api_impl
  17 from nominatim.config import Configuration
  18
  19 class HTTPNominatimError(Exception):
  20     """ A special exception class for errors raised during processing.
  21     """
  22     def __init__(self, msg: str, status: int, content_type: str) -> None:
  23         self.msg = msg
  24         self.status = status
  25         self.content_type = content_type
  26
  27
  28 async def nominatim_error_handler(req: Request, resp: Response, #pylint: disable=unused-argument
  29                                   exception: HTTPNominatimError,
  30                                   _: Any) -> None:
  31     """ Special error handler that passes message and content type as
  32         per exception info.
  33     """
  34     resp.status = exception.status
  35     resp.text = exception.msg
  36     resp.content_type = exception.content_type
  37
  38
  39 class ParamWrapper(api_impl.ASGIAdaptor):
  40     """ Adaptor class for server glue to Falcon framework.
  41     """
  42
  43     def __init__(self, req: Request, resp: Response,
  44                  config: Configuration) -> None:
  45         self.request = req
  46         self.response = resp
  47         self._config = config
  48
  49
  50     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  51         return cast(Optional[str], self.request.get_param(name, default=default))
  52
  53
  54     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  55         return cast(Optional[str], self.request.get_header(name, default=default))
  56
  57
  58     def error(self, msg: str, status: int = 400) -> HTTPNominatimError:
  59         return HTTPNominatimError(msg, status, self.content_type)
  60
  61
  62     def create_response(self, status: int, output: str) -> None:
  63         self.response.status = status
  64         self.response.text = output
  65         self.response.content_type = self.content_type
  66
  67
  68     def config(self) -> Configuration:
  69         return self._config
  70
  71
  72 class EndpointWrapper:
  73     """ Converter for server glue endpoint functions to Falcon request handlers.
  74     """
  75
  76     def __init__(self, func: api_impl.EndpointFunc, api: NominatimAPIAsync) -> None:
  77         self.func = func
  78         self.api = api
  79
  80
  81     async def on_get(self, req: Request, resp: Response) -> None:
  82         """ Implementation of the endpoint.
  83         """
  84         await self.func(self.api, ParamWrapper(req, resp, self.api.config))
  85
  86
  87 def get_application(project_dir: Path,
  88                     environ: Optional[Mapping[str, str]] = None) -> App:
  89     """ Create a Nominatim Falcon ASGI application.
  90     """
  91     api = NominatimAPIAsync(project_dir, environ)
  92
  93     app = App(cors_enable=api.config.get_bool('CORS_NOACCESSCONTROL'))
  94     app.add_error_handler(HTTPNominatimError, nominatim_error_handler)
  95
  96     legacy_urls = api.config.get_bool('SERVE_LEGACY_URLS')
  97     for name, func in api_impl.ROUTES:
  98         endpoint = EndpointWrapper(func, api)
  99         app.add_route(f"/{name}", endpoint)
 100         if legacy_urls:
 101             app.add_route(f"/{name}.php", endpoint)
 102
 103     return app
 104
 105
 106 def run_wsgi() -> App:
 107     """ Entry point for uvicorn.
 108
 109         Make sure uvicorn is run from the project directory.
 110     """
 111     return get_application(Path('.'))