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
  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
  21 class ParamWrapper(api_impl.ASGIAdaptor):
  22     """ Adaptor class for server glue to Falcon framework.
  23     """
  24
  25     def __init__(self, req: Request, resp: Response,
  26                  config: Configuration) -> None:
  27         self.request = req
  28         self.response = resp
  29         self._config = config
  30
  31
  32     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  33         return cast(Optional[str], self.request.get_param(name, default=default))
  34
  35
  36     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  37         return cast(Optional[str], self.request.get_header(name, default=default))
  38
  39
  40     def error(self, msg: str, status: int = 400) -> falcon.HTTPError:
  41         if status == 400:
  42             return falcon.HTTPBadRequest(description=msg)
  43         if status == 404:
  44             return falcon.HTTPNotFound(description=msg)
  45
  46         return falcon.HTTPError(status, description=msg)
  47
  48
  49     def create_response(self, status: int, output: str) -> None:
  50         self.response.status = status
  51         self.response.text = output
  52         self.response.content_type = self.content_type
  53
  54
  55     def config(self) -> Configuration:
  56         return self._config
  57
  58
  59 class EndpointWrapper:
  60     """ Converter for server glue endpoint functions to Falcon request handlers.
  61     """
  62
  63     def __init__(self, func: api_impl.EndpointFunc, api: NominatimAPIAsync) -> None:
  64         self.func = func
  65         self.api = api
  66
  67
  68     async def on_get(self, req: Request, resp: Response) -> None:
  69         """ Implementation of the endpoint.
  70         """
  71         await self.func(self.api, ParamWrapper(req, resp, self.api.config))
  72
  73
  74 def get_application(project_dir: Path,
  75                     environ: Optional[Mapping[str, str]] = None) -> App:
  76     """ Create a Nominatim Falcon ASGI application.
  77     """
  78     api = NominatimAPIAsync(project_dir, environ)
  79
  80     app = App(cors_enable=api.config.get_bool('CORS_NOACCESSCONTROL'))
  81
  82     legacy_urls = api.config.get_bool('SERVE_LEGACY_URLS')
  83     for name, func in api_impl.ROUTES:
  84         endpoint = EndpointWrapper(func, api)
  85         app.add_route(f"/{name}", endpoint)
  86         if legacy_urls:
  87             app.add_route(f"/{name}.php", endpoint)
  88
  89     return app