from typing import Optional, Any, Type, Callable
import abc
+from nominatim.config import Configuration
import nominatim.api as napi
from nominatim.api.v1.format import dispatch as formatting
@abc.abstractmethod
- def error(self, msg: str) -> Exception:
+ def error(self, msg: str, status: int = 400) -> Exception:
""" Construct an appropriate exception from the given error message.
- The exception must result in a HTTP 400 error.
+ The exception must result in a HTTP error with the given status.
"""
"""
+ @abc.abstractmethod
+ def config(self) -> Configuration:
+ """ Return the current configuration object.
+ """
+
+
def build_response(self, output: str, media_type: str, status: int = 200) -> Any:
""" Create a response from the given output. Wraps a JSONP function
around the response, if necessary.
return value != '0'
+ def get_accepted_languages(self) -> str:
+ """ Return the accepted langauges.
+ """
+ return self.get('accept-language')\
+ or self.get_header('http_accept_language')\
+ or self.config().DEFAULT_LANGUAGE
+
+
def parse_format(params: ASGIAdaptor, result_type: Type[Any], default: str) -> str:
""" Get and check the 'format' parameter and prepare the formatter.
`fmtter` is a formatter and `default` the
return params.build_response(formatting.format_result(result, fmt, {}), fmt,
status=status_code)
+
+async def details_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
+ """ Server glue for /details endpoint. See API docs for details.
+ """
+ place_id = params.get_int('place_id', 0)
+ place: napi.PlaceRef
+ if place_id:
+ place = napi.PlaceID(place_id)
+ else:
+ osmtype = params.get('osmtype')
+ if osmtype is None:
+ raise params.error("Missing ID parameter 'place_id' or 'osmtype'.")
+ place = napi.OsmID(osmtype, params.get_int('osmid'), params.get('class'))
+
+ details = napi.LookupDetails(address_details=params.get_bool('addressdetails', False),
+ linked_places=params.get_bool('linkedplaces', False),
+ parented_places=params.get_bool('hierarchy', False),
+ keywords=params.get_bool('keywords', False))
+
+ if params.get_bool('polygon_geojson', False):
+ details.geometry_output = napi.GeometryFormat.GEOJSON
+
+ locales = napi.Locales.from_accept_languages(params.get_accepted_languages())
+ print(locales.languages)
+
+ result = await api.lookup(place, details)
+
+ if result is None:
+ raise params.error('No place with that OSM ID found.', status=404)
+
+ output = formatting.format_result(
+ result,
+ 'details-json',
+ {'locales': locales,
+ 'group_hierarchy': params.get_bool('group_hierarchy', False),
+ 'icon_base_url': params.config().MAPICON_URL})
+
+ return params.build_response(output, 'json')
+
+
EndpointFunc = Callable[[napi.NominatimAPIAsync, ASGIAdaptor], Any]
ROUTES = [
- ('status', status_endpoint)
+ ('status', status_endpoint),
+ ('details', details_endpoint)
]
from nominatim.api import NominatimAPIAsync
import nominatim.api.v1 as api_impl
+from nominatim.config import Configuration
class ParamWrapper(api_impl.ASGIAdaptor):
""" Adaptor class for server glue to Falcon framework.
"""
- def __init__(self, req: Request, resp: Response) -> None:
+ def __init__(self, req: Request, resp: Response,
+ config: Configuration) -> None:
self.request = req
self.response = resp
+ self._config = config
def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
return cast(Optional[str], self.request.get_header(name, default=default))
- def error(self, msg: str) -> falcon.HTTPBadRequest:
- return falcon.HTTPBadRequest(description=msg)
+ def error(self, msg: str, status: int = 400) -> falcon.HTTPError:
+ if status == 400:
+ return falcon.HTTPBadRequest(description=msg)
+ if status == 404:
+ return falcon.HTTPNotFound(description=msg)
+
+ return falcon.HTTPError(status, description=msg)
def create_response(self, status: int, output: str, content_type: str) -> None:
self.response.content_type = content_type
+ def config(self) -> Configuration:
+ return self._config
+
+
class EndpointWrapper:
""" Converter for server glue endpoint functions to Falcon request handlers.
"""
async def on_get(self, req: Request, resp: Response) -> None:
""" Implementation of the endpoint.
"""
- await self.func(self.api, ParamWrapper(req, resp))
+ await self.func(self.api, ParamWrapper(req, resp, self.api.config))
def get_application(project_dir: Path,
from nominatim.api import NominatimAPIAsync
import nominatim.api.v1 as api_impl
+from nominatim.config import Configuration
class ParamWrapper(api_impl.ASGIAdaptor):
""" Adaptor class for server glue to Sanic framework.
return cast(Optional[str], self.request.headers.get(name, default))
- def error(self, msg: str) -> SanicException:
- return SanicException(msg, status_code=400)
+ def error(self, msg: str, status: int = 400) -> SanicException:
+ return SanicException(msg, status_code=status)
def create_response(self, status: int, output: str,
return TextResponse(output, status=status, content_type=content_type)
+ def config(self) -> Configuration:
+ return cast(Configuration, self.request.app.ctx.api.config)
+
+
def _wrap_endpoint(func: api_impl.EndpointFunc)\
-> Callable[[Request], Coroutine[Any, Any, HTTPResponse]]:
async def _callback(request: Request) -> HTTPResponse:
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
-from nominatim.config import Configuration
from nominatim.api import NominatimAPIAsync
import nominatim.api.v1 as api_impl
+from nominatim.config import Configuration
class ParamWrapper(api_impl.ASGIAdaptor):
""" Adaptor class for server glue to Starlette framework.
return self.request.headers.get(name, default)
- def error(self, msg: str) -> HTTPException:
- return HTTPException(400, detail=msg)
+ def error(self, msg: str, status: int = 400) -> HTTPException:
+ return HTTPException(status, detail=msg)
def create_response(self, status: int, output: str, content_type: str) -> Response:
return Response(output, status_code=status, media_type=content_type)
+ def config(self) -> Configuration:
+ return cast(Configuration, self.request.app.state.API.config)
+
+
def _wrap_endpoint(func: api_impl.EndpointFunc)\
-> Callable[[Request], Coroutine[Any, Any, Response]]:
async def _callback(request: Request) -> Response:
from asgi_lifespan import LifespanManager
import httpx
- async def _request(endpoint, params, project_dir, environ):
+ async def _request(endpoint, params, project_dir, environ, http_headers):
app = nominatim.server.starlette.server.get_application(project_dir, environ)
async with LifespanManager(app):
async with httpx.AsyncClient(app=app, base_url="http://nominatim.test") as client:
- response = await client.get(f"/{endpoint}", params=params)
+ response = await client.get(f"/{endpoint}", params=params,
+ headers=http_headers)
return response.text, response.status_code
def create_api_request_func_sanic(self):
import nominatim.server.sanic.server
- async def _request(endpoint, params, project_dir, environ):
+ async def _request(endpoint, params, project_dir, environ, http_headers):
app = nominatim.server.sanic.server.get_application(project_dir, environ)
- _, response = await app.asgi_client.get(f"/{endpoint}", params=params)
+ _, response = await app.asgi_client.get(f"/{endpoint}", params=params,
+ headers=http_headers)
return response.text, response.status_code
import nominatim.server.falcon.server
import falcon.testing
- async def _request(endpoint, params, project_dir, environ):
+ async def _request(endpoint, params, project_dir, environ, http_headers):
app = nominatim.server.falcon.server.get_application(project_dir, environ)
async with falcon.testing.ASGIConductor(app) as conductor:
- response = await conductor.get(f"/{endpoint}", params=params)
+ response = await conductor.get(f"/{endpoint}", params=params,
+ headers=http_headers)
return response.text, response.status_code
return asyncio.run(context.nominatim.api_engine(endpoint, params,
Path(context.nominatim.website_dir.name),
- context.nominatim.test_env))
+ context.nominatim.test_env,
+ getattr(context, 'http_headers', {})))
projects / nominatim.git / commitdiff