[mypy-dotenv.*]
ignore_missing_imports = True
+
+[mypy-falcon.*]
+ignore_missing_imports = True
[MASTER]
-extension-pkg-whitelist=osmium
+extension-pkg-whitelist=osmium,falcon
ignored-modules=icu,datrie
[MESSAGES CONTROL]
"""
Implementation of classes for API access via libraries.
"""
-from typing import Mapping, Optional, TypeVar, Callable, Any
-import functools
+from typing import Mapping, Optional
import asyncio
from pathlib import Path
from sqlalchemy.engine.url import URL
from sqlalchemy.ext.asyncio import create_async_engine
-from nominatim.typing import StrPath
from nominatim.config import Configuration
from nominatim.apicmd.status import get_status, StatusResult
def status(self) -> StatusResult:
+ """ Return the status of the database.
+ """
return asyncio.get_event_loop().run_until_complete(self.async_api.status())
database administration and querying.
"""
from typing import Optional, Any, List, Union
+import importlib
import logging
import os
import sys
port = 8088
if args.engine == 'sanic':
- import nominatim.server.sanic.server
+ server_module = importlib.import_module('nominatim.server.sanic.server')
- app = nominatim.server.sanic.server.get_application(args.project_dir)
+ app = server_module.get_application(args.project_dir)
app.run(host=host, port=port, debug=True)
else:
- import uvicorn
+ import uvicorn # pylint: disable=import-outside-toplevel
if args.engine == 'falcon':
- import nominatim.server.falcon.server as server_module
+ server_module = importlib.import_module('nominatim.server.falcon.server')
elif args.engine == 'starlette':
- import nominatim.server.starlette.server as server_module
+ server_module = importlib.import_module('nominatim.server.starlette.server')
app = server_module.get_application(args.project_dir)
uvicorn.run(app, host=host, port=port)
"""
Server implementation using the falcon webserver framework.
"""
+from typing import Type, Any
from pathlib import Path
+import falcon
import falcon.asgi
from nominatim.api import NominatimAPIAsync
}
class NominatimV1:
+ """ Implementation of V1 version of the Nominatim API.
+ """
- def __init__(self, project_dir: Path):
+ def __init__(self, project_dir: Path) -> None:
self.api = NominatimAPIAsync(project_dir)
self.formatters = {}
for rtype in (StatusResult, ):
self.formatters[rtype] = formatting.create(rtype)
- def parse_format(self, req, rtype, default):
+
+ def parse_format(self, req: falcon.asgi.Request, rtype: Type[Any], default: str) -> None:
+ """ Get and check the 'format' parameter and prepare the formatter.
+ `rtype` describes the expected return type and `default` the
+ format value to assume when no parameter is present.
+ """
req.context.format = req.get_param('format', default=default)
req.context.formatter = self.formatters[rtype]
', '.join(req.context.formatter.list_formats()))
- def format_response(self, req, resp, result):
+ def format_response(self, req: falcon.asgi.Request, resp: falcon.asgi.Response,
+ result: Any) -> None:
+ """ Render response into a string according to the formatter
+ set in `parse_format()`.
+ """
resp.text = req.context.formatter.format(result, req.context.format)
resp.content_type = CONTENT_TYPE.get(req.context.format, falcon.MEDIA_JSON)
- async def on_get_status(self, req, resp):
+ async def on_get_status(self, req: falcon.asgi.Request, resp: falcon.asgi.Response) -> None:
+ """ Implementation of status endpoint.
+ """
self.parse_format(req, StatusResult, 'text')
result = await self.api.status()
def get_application(project_dir: Path) -> falcon.asgi.App:
+ """ Create a Nominatim falcon ASGI application.
+ """
app = falcon.asgi.App()
api = NominatimV1(project_dir)
"""
Server implementation using the sanic webserver framework.
"""
+from typing import Any, Optional
from pathlib import Path
import sanic
'xml': 'text/xml; charset=utf-8'
}
-def usage_error(msg):
+def usage_error(msg: str) -> sanic.HTTPResponse:
+ """ Format the response for an error with the query parameters.
+ """
return sanic.response.text(msg, status=400)
-def api_response(request, result):
+def api_response(request: sanic.Request, result: Any) -> sanic.HTTPResponse:
+ """ Render a response from the query results using the configured
+ formatter.
+ """
body = request.ctx.formatter.format(result, request.ctx.format)
return sanic.response.text(body,
- content_type=CONTENT_TYPE.get(request.ctx.format, 'application/json'))
-
-
-@api.on_request
-async def extract_format(request):
+ content_type=CONTENT_TYPE.get(request.ctx.format,
+ 'application/json'))
+
+
+@api.on_request # type: ignore[misc]
+async def extract_format(request: sanic.Request) -> Optional[sanic.HTTPResponse]:
+ """ Get and check the 'format' parameter and prepare the formatter.
+ `ctx.result_type` describes the expected return type and
+ `ctx.default_format` the format value to assume when no parameter
+ is present.
+ """
+ assert request.route is not None
request.ctx.formatter = request.app.ctx.formatters[request.route.ctx.result_type]
request.ctx.format = request.args.get('format', request.route.ctx.default_format)
return usage_error("Parameter 'format' must be one of: " +
', '.join(request.ctx.formatter.list_formats()))
+ return None
+
@api.get('/status', ctx_result_type=StatusResult, ctx_default_format='text')
-async def status(request):
+async def status(request: sanic.Request) -> sanic.HTTPResponse:
+ """ Implementation of status endpoint.
+ """
return api_response(request,await request.app.ctx.api.status())
def get_application(project_dir: Path) -> sanic.Sanic:
+ """ Create a Nominatim sanic ASGI application.
+ """
app = sanic.Sanic("NominatimInstance")
app.ctx.api = NominatimAPIAsync(project_dir)
app.blueprint(api)
return app
-
-
"""
Server implementation using the starlette webserver framework.
"""
+from typing import Any, Type
from pathlib import Path
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.exceptions import HTTPException
from starlette.responses import Response
+from starlette.requests import Request
from nominatim.api import NominatimAPIAsync
from nominatim.apicmd.status import StatusResult
}
-def parse_format(request, rtype, default):
+def parse_format(request: Request, rtype: Type[Any], default: str) -> None:
+ """ Get and check the 'format' parameter and prepare the formatter.
+ `rtype` describes the expected return type and `default` the
+ format value to assume when no parameter is present.
+ """
fmt = request.query_params.get('format', default=default)
fmtter = FORMATTERS[rtype]
request.state.formatter = fmtter
-def format_response(request, result):
+def format_response(request: Request, result: Any) -> Response:
+ """ Render response into a string according to the formatter
+ set in `parse_format()`.
+ """
fmt = request.state.format
return Response(request.state.formatter.format(result, fmt),
media_type=CONTENT_TYPE.get(fmt, 'application/json'))
-async def on_status(request):
+async def on_status(request: Request) -> Response:
+ """ Implementation of status endpoint.
+ """
parse_format(request, StatusResult, 'text')
result = await request.app.state.API.status()
return format_response(request, result)
]
def get_application(project_dir: Path) -> Starlette:
+ """ Create a Nominatim falcon ASGI application.
+ """
app = Starlette(debug=True, routes=V1_ROUTES)
app.state.API = NominatimAPIAsync(project_dir)
projects / nominatim.git / commitdiff