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
  19
  20 class ParamWrapper(api_impl.ASGIAdaptor):
  21     """ Adaptor class for server glue to Falcon framework.
  22     """
  23
  24     def __init__(self, req: Request, resp: Response) -> None:
  25         self.request = req
  26         self.response = resp
  27
  28
  29     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  30         return cast(Optional[str], self.request.get_param(name, default=default))
  31
  32
  33     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  34         return cast(Optional[str], self.request.get_header(name, default=default))
  35
  36
  37     def error(self, msg: str) -> falcon.HTTPBadRequest:
  38         return falcon.HTTPBadRequest(description=msg)
  39
  40
  41     def create_response(self, status: int, output: str, content_type: str) -> None:
  42         self.response.status = status
  43         self.response.text = output
  44         self.response.content_type = content_type
  45
  46
  47 class EndpointWrapper:
  48     """ Converter for server glue endpoint functions to Falcon request handlers.
  49     """
  50
  51     def __init__(self, func: api_impl.EndpointFunc, api: NominatimAPIAsync) -> None:
  52         self.func = func
  53         self.api = api
  54
  55
  56     async def on_get(self, req: Request, resp: Response) -> None:
  57         """ Implementation of the endpoint.
  58         """
  59         await self.func(self.api, ParamWrapper(req, resp))
  60
  61
  62 def get_application(project_dir: Path,
  63                     environ: Optional[Mapping[str, str]] = None) -> App:
  64     """ Create a Nominatim Falcon ASGI application.
  65     """
  66     api = NominatimAPIAsync(project_dir, environ)
  67
  68     app = App()
  69     for name, func in api_impl.ROUTES:
  70         app.add_route('/' + name, EndpointWrapper(func, api))
  71
  72     return app