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