mirror of
https://github.com/osm-search/Nominatim.git
synced 2024-08-17 07:00:34 +03:00
fix liniting issues and add type annotations
This commit is contained in:
parent
1adb0a9886
commit
d7bc846c3c
@ -12,3 +12,6 @@ ignore_missing_imports = True
|
||||
|
||||
[mypy-dotenv.*]
|
||||
ignore_missing_imports = True
|
||||
|
||||
[mypy-falcon.*]
|
||||
ignore_missing_imports = True
|
||||
|
@ -1,6 +1,6 @@
|
||||
[MASTER]
|
||||
|
||||
extension-pkg-whitelist=osmium
|
||||
extension-pkg-whitelist=osmium,falcon
|
||||
ignored-modules=icu,datrie
|
||||
|
||||
[MESSAGES CONTROL]
|
||||
|
@ -7,15 +7,13 @@
|
||||
"""
|
||||
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
|
||||
|
||||
@ -56,4 +54,6 @@ class NominatimAPI:
|
||||
|
||||
|
||||
def status(self) -> StatusResult:
|
||||
""" Return the status of the database.
|
||||
"""
|
||||
return asyncio.get_event_loop().run_until_complete(self.async_api.status())
|
||||
|
@ -9,6 +9,7 @@ Command-line interface to the Nominatim functions for import, update,
|
||||
database administration and querying.
|
||||
"""
|
||||
from typing import Optional, Any, List, Union
|
||||
import importlib
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
@ -232,17 +233,17 @@ class AdminServe:
|
||||
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)
|
||||
|
@ -7,8 +7,10 @@
|
||||
"""
|
||||
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
|
||||
@ -21,15 +23,22 @@ CONTENT_TYPE = {
|
||||
}
|
||||
|
||||
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]
|
||||
|
||||
@ -39,12 +48,18 @@ class NominatimV1:
|
||||
', '.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()
|
||||
@ -53,6 +68,8 @@ class NominatimV1:
|
||||
|
||||
|
||||
def get_application(project_dir: Path) -> falcon.asgi.App:
|
||||
""" Create a Nominatim falcon ASGI application.
|
||||
"""
|
||||
app = falcon.asgi.App()
|
||||
|
||||
api = NominatimV1(project_dir)
|
||||
|
@ -7,6 +7,7 @@
|
||||
"""
|
||||
Server implementation using the sanic webserver framework.
|
||||
"""
|
||||
from typing import Any, Optional
|
||||
from pathlib import Path
|
||||
|
||||
import sanic
|
||||
@ -22,18 +23,30 @@ CONTENT_TYPE = {
|
||||
'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'))
|
||||
content_type=CONTENT_TYPE.get(request.ctx.format,
|
||||
'application/json'))
|
||||
|
||||
|
||||
@api.on_request
|
||||
async def extract_format(request):
|
||||
@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)
|
||||
@ -41,13 +54,19 @@ async def extract_format(request):
|
||||
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)
|
||||
@ -58,5 +77,3 @@ def get_application(project_dir: Path) -> sanic.Sanic:
|
||||
app.blueprint(api)
|
||||
|
||||
return app
|
||||
|
||||
|
||||
|
@ -7,12 +7,14 @@
|
||||
"""
|
||||
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
|
||||
@ -28,7 +30,11 @@ FORMATTERS = {
|
||||
}
|
||||
|
||||
|
||||
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]
|
||||
|
||||
@ -40,13 +46,18 @@ def parse_format(request, rtype, default):
|
||||
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)
|
||||
@ -57,6 +68,8 @@ V1_ROUTES = [
|
||||
]
|
||||
|
||||
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)
|
||||
|
Loading…
Reference in New Issue
Block a user