nominatim/server/starlette/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 starlette webserver framework.
   9 """
  10 from typing import Any, Optional, Mapping, Callable, cast, Coroutine
  11 from pathlib import Path
  12
  13 from starlette.applications import Starlette
  14 from starlette.routing import Route
  15 from starlette.exceptions import HTTPException
  16 from starlette.responses import Response
  17 from starlette.requests import Request
  18 from starlette.middleware import Middleware
  19 from starlette.middleware.cors import CORSMiddleware
  20
  21 from nominatim.api import NominatimAPIAsync
  22 import nominatim.api.v1 as api_impl
  23 from nominatim.config import Configuration
  24
  25 class ParamWrapper(api_impl.ASGIAdaptor):
  26     """ Adaptor class for server glue to Starlette framework.
  27     """
  28
  29     def __init__(self, request: Request) -> None:
  30         self.request = request
  31
  32
  33     def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
  34         return self.request.query_params.get(name, default=default)
  35
  36
  37     def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
  38         return self.request.headers.get(name, default)
  39
  40
  41     def error(self, msg: str, status: int = 400) -> HTTPException:
  42         return HTTPException(status, detail=msg)
  43
  44
  45     def create_response(self, status: int, output: str, content_type: str) -> Response:
  46         return Response(output, status_code=status, media_type=content_type)
  47
  48
  49     def config(self) -> Configuration:
  50         return cast(Configuration, self.request.app.state.API.config)
  51
  52
  53 def _wrap_endpoint(func: api_impl.EndpointFunc)\
  54         -> Callable[[Request], Coroutine[Any, Any, Response]]:
  55     async def _callback(request: Request) -> Response:
  56         return cast(Response, await func(request.app.state.API, ParamWrapper(request)))
  57
  58     return _callback
  59
  60
  61 def get_application(project_dir: Path,
  62                     environ: Optional[Mapping[str, str]] = None) -> Starlette:
  63     """ Create a Nominatim falcon ASGI application.
  64     """
  65     config = Configuration(project_dir, environ)
  66
  67     routes = []
  68     legacy_urls = config.get_bool('SERVE_LEGACY_URLS')
  69     for name, func in api_impl.ROUTES:
  70         endpoint = _wrap_endpoint(func)
  71         routes.append(Route(f"/{name}", endpoint=endpoint))
  72         if legacy_urls:
  73             routes.append(Route(f"/{name}.php", endpoint=endpoint))
  74
  75     middleware = []
  76     if config.get_bool('CORS_NOACCESSCONTROL'):
  77         middleware.append(Middleware(CORSMiddleware, allow_origins=['*']))
  78
  79     app = Starlette(debug=True, routes=routes, middleware=middleware)
  80
  81     app.state.API = NominatimAPIAsync(project_dir, environ)
  82
  83     return app