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