if: matrix.flavour == 'oldstuff'
- name: Install Python webservers
- run: pip3 install falcon sanic sanic-testing sanic-cors starlette
+ run: pip3 install falcon starlette
- name: Install latest pylint
run: pip3 install -U pylint asgi_lifespan
* one of the following web frameworks:
* [falcon](https://falconframework.org/) (3.0+)
- * [sanic](https://sanic.dev) and (optionally) [sanic-cors](https://github.com/ashleysommer/sanic-cors)
* [starlette](https://www.starlette.io/)
- * [uvicorn](https://www.uvicorn.org/) (only with falcon and starlette framworks)
+ * [uvicorn](https://www.uvicorn.org/)
For dependencies for running tests and building documentation, see
the [Development section](../develop/Development-Environment.md).
For testing the Python search frontend, you need to install extra dependencies
depending on your choice of webserver framework:
-* [sanic-testing](https://sanic.dev/en/plugins/sanic-testing/getting-started.html) (sanic only)
* [httpx](https://www.python-httpx.org/) (starlette only)
* [asgi-lifespan](https://github.com/florimondmanca/asgi-lifespan) (starlette only)
pip3 install --user behave mkdocs mkdocstrings pytest pytest-asyncio pylint \
mypy types-PyYAML types-jinja2 types-psycopg2 types-psutil \
types-ujson types-requests types-Pygments typing-extensions\
- sanic-testing httpx asgi-lifespan
+ httpx asgi-lifespan
```
The `mkdocs` executable will be located in `.local/bin`. You may have to add
and (repl.ttype != qmod.TokenType.HOUSENUMBER
or len(tlist.tokens[0].lookup_word) > 4):
repl.add_penalty(0.39)
- elif tlist.ttype == qmod.TokenType.HOUSENUMBER:
+ elif tlist.ttype == qmod.TokenType.HOUSENUMBER \
+ and len(tlist.tokens[0].lookup_word) <= 3:
if any(c.isdigit() for c in tlist.tokens[0].lookup_word):
for repl in node.starting:
- if repl.end == tlist.end and repl.ttype != qmod.TokenType.HOUSENUMBER \
- and (repl.ttype != qmod.TokenType.HOUSENUMBER
- or len(tlist.tokens[0].lookup_word) <= 3):
+ if repl.end == tlist.end and repl.ttype != qmod.TokenType.HOUSENUMBER:
repl.add_penalty(0.5 - tlist.tokens[0].penalty)
elif tlist.ttype not in (qmod.TokenType.COUNTRY, qmod.TokenType.PARTIAL):
norm = parts[i].normalized
and (repl.ttype != qmod.TokenType.HOUSENUMBER
or len(tlist.tokens[0].lookup_word) > 4):
repl.add_penalty(0.39)
- elif tlist.ttype == qmod.TokenType.HOUSENUMBER:
+ elif tlist.ttype == qmod.TokenType.HOUSENUMBER \
+ and len(tlist.tokens[0].lookup_word) <= 3:
if any(c.isdigit() for c in tlist.tokens[0].lookup_word):
for repl in node.starting:
- if repl.end == tlist.end and repl.ttype != qmod.TokenType.HOUSENUMBER \
- and (repl.ttype != qmod.TokenType.HOUSENUMBER
- or len(tlist.tokens[0].lookup_word) <= 3):
+ if repl.end == tlist.end and repl.ttype != qmod.TokenType.HOUSENUMBER:
repl.add_penalty(0.5 - tlist.tokens[0].penalty)
"""
parsed = SearchDetails.from_kwargs(details)
if parsed.geometry_output != GeometryFormat.NONE:
- if parsed.geometry_output & GeometryFormat.GEOJSON:
+ if GeometryFormat.GEOJSON in parsed.geometry_output:
queryparts['polygon_geojson'] = '1'
- if parsed.geometry_output & GeometryFormat.KML:
+ if GeometryFormat.KML in parsed.geometry_output:
queryparts['polygon_kml'] = '1'
- if parsed.geometry_output & GeometryFormat.SVG:
+ if GeometryFormat.SVG in parsed.geometry_output:
queryparts['polygon_svg'] = '1'
- if parsed.geometry_output & GeometryFormat.TEXT:
+ if GeometryFormat.TEXT in parsed.geometry_output:
queryparts['polygon_text'] = '1'
if parsed.address_details:
queryparts['addressdetails'] = '1'
group.add_argument('--server', default='127.0.0.1:8088',
help='The address the server will listen to.')
group.add_argument('--engine', default='php',
- choices=('php', 'sanic', 'falcon', 'starlette'),
+ choices=('php', 'falcon', 'starlette'),
help='Webserver framework to run. (default: php)')
if args.engine == 'php':
run_php_server(args.server, args.project_dir / 'website')
else:
+ import uvicorn # pylint: disable=import-outside-toplevel
server_info = args.server.split(':', 1)
host = server_info[0]
if len(server_info) > 1:
else:
port = 8088
- if args.engine == 'sanic':
- server_module = importlib.import_module('nominatim.server.sanic.server')
+ server_module = importlib.import_module(f'nominatim.server.{args.engine}.server')
- app = server_module.get_application(args.project_dir)
- app.run(host=host, port=port, debug=True, single_process=True)
- else:
- import uvicorn # pylint: disable=import-outside-toplevel
-
- if args.engine == 'falcon':
- server_module = importlib.import_module('nominatim.server.falcon.server')
- elif args.engine == 'starlette':
- server_module = importlib.import_module('nominatim.server.starlette.server')
-
- app = server_module.get_application(args.project_dir)
- uvicorn.run(app, host=host, port=port)
+ app = server_module.get_application(args.project_dir)
+ uvicorn.run(app, host=host, port=port)
return 0
+++ /dev/null
-# SPDX-License-Identifier: GPL-2.0-only
-#
-# This file is part of Nominatim. (https://nominatim.org)
-#
-# Copyright (C) 2023 by the Nominatim developer community.
-# For a full list of authors see the git log.
-"""
-Server implementation using the sanic webserver framework.
-"""
-from typing import Any, Optional, Mapping, Callable, cast, Coroutine
-from pathlib import Path
-
-from sanic import Request, HTTPResponse, Sanic
-from sanic.exceptions import SanicException
-from sanic.response import text as TextResponse
-
-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.
- """
-
- def __init__(self, request: Request) -> None:
- self.request = request
-
-
- def get(self, name: str, default: Optional[str] = None) -> Optional[str]:
- return cast(Optional[str], self.request.args.get(name, default))
-
-
- def get_header(self, name: str, default: Optional[str] = None) -> Optional[str]:
- return cast(Optional[str], self.request.headers.get(name, default))
-
-
- def error(self, msg: str, status: int = 400) -> SanicException:
- exception = SanicException(msg, status_code=status)
-
- return exception
-
-
- def create_response(self, status: int, output: str) -> HTTPResponse:
- return TextResponse(output, status=status, content_type=self.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:
- return cast(HTTPResponse, await func(request.app.ctx.api, ParamWrapper(request)))
-
- return _callback
-
-
-def get_application(project_dir: Path,
- environ: Optional[Mapping[str, str]] = None) -> Sanic:
- """ Create a Nominatim sanic ASGI application.
- """
- app = Sanic("NominatimInstance")
-
- app.ctx.api = NominatimAPIAsync(project_dir, environ)
-
- if app.ctx.api.config.get_bool('CORS_NOACCESSCONTROL'):
- from sanic_cors import CORS # pylint: disable=import-outside-toplevel
- CORS(app)
-
- legacy_urls = app.ctx.api.config.get_bool('SERVE_LEGACY_URLS')
- for name, func in api_impl.ROUTES:
- endpoint = _wrap_endpoint(func)
- app.add_route(endpoint, f"/{name}", name=f"v1_{name}_simple")
- if legacy_urls:
- app.add_route(endpoint, f"/{name}.php", name=f"v1_{name}_legacy")
-
- return app
return _request
- def create_api_request_func_sanic(self):
- import nominatim.server.sanic.server
-
- 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,
- headers=http_headers)
-
- return response.text, response.status_code
-
- return _request
-
-
def create_api_request_func_falcon(self):
import nominatim.server.falcon.server
import falcon.testing
assert query.num_token_slots() == 1
- torder = [(tl.tokens[0].penalty, tl.ttype) for tl in query.nodes[0].starting]
+ torder = [(tl.tokens[0].penalty, tl.ttype.name) for tl in query.nodes[0].starting]
torder.sort()
- assert [t[1] for t in torder] == [TokenType[o] for o in order]
+ assert [t[1] for t in torder] == order
@pytest.mark.asyncio
async def test_category_words_only_at_beginning(conn):
assert query.num_token_slots() == 1
- torder = [(tl.tokens[0].penalty, tl.ttype) for tl in query.nodes[0].starting]
- print(query.nodes[0].starting)
+ torder = [(tl.tokens[0].penalty, tl.ttype.name) for tl in query.nodes[0].starting]
torder.sort()
- assert [t[1] for t in torder] == [TokenType[o] for o in order]
+ assert [t[1] for t in torder] == order
@pytest.mark.asyncio
assert func.called == 1
-def test_cli_serve_sanic(cli_call, mock_func_factory):
- mod = pytest.importorskip("sanic")
- func = mock_func_factory(mod.Sanic, "run")
-
- cli_call('serve', '--engine', 'sanic') == 0
-
- assert func.called == 1
-
-
def test_cli_serve_starlette_custom_server(cli_call, mock_func_factory):
pytest.importorskip("starlette")
mod = pytest.importorskip("uvicorn")
projects / nominatim.git / commitdiff