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, Type, Optional, Mapping
  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
  19 from nominatim.api import NominatimAPIAsync, StatusResult
  20 import nominatim.api.v1 as api_impl
  21
  22 CONTENT_TYPE = {
  23   'text': 'text/plain; charset=utf-8',
  24   'xml': 'text/xml; charset=utf-8'
  25 }
  26
  27 def parse_format(request: Request, rtype: Type[Any], default: str) -> None:
  28     """ Get and check the 'format' parameter and prepare the formatter.
  29         `rtype` describes the expected return type and `default` the
  30         format value to assume when no parameter is present.
  31     """
  32     fmt = request.query_params.get('format', default=default)
  33
  34     if not api_impl.supports_format(rtype, fmt):
  35         raise HTTPException(400, detail="Parameter 'format' must be one of: " +
  36                                         ', '.join(api_impl.list_formats(rtype)))
  37
  38     request.state.format = fmt
  39
  40
  41 def format_response(request: Request, result: Any) -> Response:
  42     """ Render response into a string according.
  43     """
  44     fmt = request.state.format
  45     return Response(api_impl.format_result(result, fmt),
  46                     media_type=CONTENT_TYPE.get(fmt, 'application/json'))
  47
  48
  49 async def on_status(request: Request) -> Response:
  50     """ Implementation of status endpoint.
  51     """
  52     parse_format(request, StatusResult, 'text')
  53     result = await request.app.state.API.status()
  54     response = format_response(request, result)
  55
  56     if request.state.format == 'text' and result.status:
  57         response.status_code = 500
  58
  59     return response
  60
  61
  62 V1_ROUTES = [
  63     Route('/status', endpoint=on_status)
  64 ]
  65
  66 def get_application(project_dir: Path,
  67                     environ: Optional[Mapping[str, str]] = None) -> Starlette:
  68     """ Create a Nominatim falcon ASGI application.
  69     """
  70     app = Starlette(debug=True, routes=V1_ROUTES)
  71
  72     app.state.API = NominatimAPIAsync(project_dir, environ)
  73
  74     return app