nominatim/server/sanic/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 sanic webserver framework.
   9 """
  10 from typing import Any, Optional, Mapping, Callable, cast, Coroutine
  11 from pathlib import Path
  12
  13 from sanic import Request, HTTPResponse, Sanic
  14 from sanic.exceptions import SanicException
  15 from sanic.response import text as TextResponse
  16
  17 from nominatim.api import NominatimAPIAsync
  18 import nominatim.api.v1 as api_impl
  19 from nominatim.config import Configuration
  20
  21 class ParamWrapper(api_impl.ASGIAdaptor):
  22     """ Adaptor class for server glue to Sanic framework.
  23     """
  24
  25     def __init__(self, request: Request) -> None:
  26         self.request = request
  27
  28
  29     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  30         return cast(Optional[str], self.request.args.get(name, default))
  31
  32
  33     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  34         return cast(Optional[str], self.request.headers.get(name, default))
  35
  36
  37     def error(self, msg: str, status: int = 400) -> SanicException:
  38         exception = SanicException(msg, status_code=status)
  39
  40         return exception
  41
  42
  43     def create_response(self, status: int, output: str) -> HTTPResponse:
  44         return TextResponse(output, status=status, content_type=self.content_type)
  45
  46
  47     def config(self) -> Configuration:
  48         return cast(Configuration, self.request.app.ctx.api.config)
  49
  50
  51 def _wrap_endpoint(func: api_impl.EndpointFunc)\
  52        -> Callable[[Request], Coroutine[Any, Any, HTTPResponse]]:
  53     async def _callback(request: Request) -> HTTPResponse:
  54         return cast(HTTPResponse, await func(request.app.ctx.api, ParamWrapper(request)))
  55
  56     return _callback
  57
  58
  59 def get_application(project_dir: Path,
  60                     environ: Optional[Mapping[str, str]] = None) -> Sanic:
  61     """ Create a Nominatim sanic ASGI application.
  62     """
  63     app = Sanic("NominatimInstance")
  64
  65     app.ctx.api = NominatimAPIAsync(project_dir, environ)
  66
  67     if app.ctx.api.config.get_bool('CORS_NOACCESSCONTROL'):
  68         from sanic_cors import CORS # pylint: disable=import-outside-toplevel
  69         CORS(app)
  70
  71     legacy_urls = app.ctx.api.config.get_bool('SERVE_LEGACY_URLS')
  72     for name, func in api_impl.ROUTES:
  73         endpoint = _wrap_endpoint(func)
  74         app.add_route(endpoint, f"/{name}", name=f"v1_{name}_simple")
  75         if legacy_urls:
  76             app.add_route(endpoint, f"/{name}.php", name=f"v1_{name}_legacy")
  77
  78     return app