nominatim/server/falcon/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 falcon webserver framework.
   9 """
  10 from typing import Type, Any, Optional, Mapping
  11 from pathlib import Path
  12
  13 import falcon
  14 import falcon.asgi
  15
  16 from nominatim.api import NominatimAPIAsync, StatusResult
  17 import nominatim.result_formatter.v1 as formatting
  18
  19 CONTENT_TYPE = {
  20   'text': falcon.MEDIA_TEXT,
  21   'xml': falcon.MEDIA_XML
  22 }
  23
  24 class NominatimV1:
  25     """ Implementation of V1 version of the Nominatim API.
  26     """
  27
  28     def __init__(self, project_dir: Path, environ: Optional[Mapping[str, str]]) -> None:
  29         self.api = NominatimAPIAsync(project_dir, environ)
  30         self.formatters = {}
  31
  32         for rtype in (StatusResult, ):
  33             self.formatters[rtype] = formatting.create(rtype)
  34
  35
  36     def parse_format(self, req: falcon.asgi.Request, rtype: Type[Any], default: str) -> None:
  37         """ Get and check the 'format' parameter and prepare the formatter.
  38             `rtype` describes the expected return type and `default` the
  39             format value to assume when no parameter is present.
  40         """
  41         req.context.format = req.get_param('format', default=default)
  42         req.context.formatter = self.formatters[rtype]
  43
  44         if not req.context.formatter.supports_format(req.context.format):
  45             raise falcon.HTTPBadRequest(
  46                 description="Parameter 'format' must be one of: " +
  47                             ', '.join(req.context.formatter.list_formats()))
  48
  49
  50     def format_response(self, req: falcon.asgi.Request, resp: falcon.asgi.Response,
  51                         result: Any) -> None:
  52         """ Render response into a string according to the formatter
  53             set in `parse_format()`.
  54         """
  55         resp.text = req.context.formatter.format(result, req.context.format)
  56         resp.content_type = CONTENT_TYPE.get(req.context.format, falcon.MEDIA_JSON)
  57
  58
  59     async def on_get_status(self, req: falcon.asgi.Request, resp: falcon.asgi.Response) -> None:
  60         """ Implementation of status endpoint.
  61         """
  62         self.parse_format(req, StatusResult, 'text')
  63
  64         result = await self.api.status()
  65
  66         self.format_response(req, resp, result)
  67         if result.status and req.context.format == 'text':
  68             resp.status = 500
  69
  70
  71 def get_application(project_dir: Path,
  72                     environ: Optional[Mapping[str, str]] = None) -> falcon.asgi.App:
  73     """ Create a Nominatim falcon ASGI application.
  74     """
  75     app = falcon.asgi.App()
  76
  77     api = NominatimV1(project_dir, environ)
  78
  79     app.add_route('/status', api, suffix='status')
  80
  81     return app