hans/Nominatim
1
0
Fork 1
mirror of https://github.com/osm-search/Nominatim.git synced 2026-02-26 11:08:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

make details API work with sqlite incl. unit tests

Browse Source
This commit is contained in:
Sarah Hoffmann
2023-10-12 15:31:20 +02:00
parent d0c91e4acf
commit 07e6c5cf69
5 changed files with 172 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
nominatim/api/lookup.py
View File

@@ -77,8 +77,8 @@ async def find_in_osmline(conn: SearchConnection, place: ntyp.PlaceRef,
sql = sql.where(t.c.osm_id == place.osm_id).limit(1) sql = sql.where(t.c.osm_id == place.osm_id).limit(1)
if place.osm_class and place.osm_class.isdigit(): if place.osm_class and place.osm_class.isdigit():
sql = sql.order_by(sa.func.greatest(0, sql = sql.order_by(sa.func.greatest(0,
sa.func.least(int(place.osm_class) - t.c.endnumber), int(place.osm_class) - t.c.endnumber,
t.c.startnumber - int(place.osm_class))) t.c.startnumber - int(place.osm_class)))
else: else:
return None return None
@@ -163,11 +163,10 @@ async def get_detailed_place(conn: SearchConnection, place: ntyp.PlaceRef,
if details.geometry_output & ntyp.GeometryFormat.GEOJSON: if details.geometry_output & ntyp.GeometryFormat.GEOJSON:
def _add_geometry(sql: SaSelect, column: SaColumn) -> SaSelect: def _add_geometry(sql: SaSelect, column: SaColumn) -> SaSelect:
return sql.add_columns(sa.literal_column(f""" return sql.add_columns(sa.func.ST_AsGeoJSON(
ST_AsGeoJSON(CASE WHEN ST_NPoints({column.name}) > 5000 sa.case((sa.func.ST_NPoints(column) > 5000,
THEN ST_SimplifyPreserveTopology({column.name}, 0.0001) sa.func.ST_SimplifyPreserveTopology(column, 0.0001)),
ELSE {column.name} END) else_=column)).label('geometry_geojson'))
""").label('geometry_geojson'))
else: else:
def _add_geometry(sql: SaSelect, column: SaColumn) -> SaSelect: def _add_geometry(sql: SaSelect, column: SaColumn) -> SaSelect:
return sql.add_columns(sa.func.ST_GeometryType(column).label('geometry_type')) return sql.add_columns(sa.func.ST_GeometryType(column).label('geometry_type'))
@@ -183,6 +182,9 @@ async def get_detailed_place(conn: SearchConnection, place: ntyp.PlaceRef,
# add missing details # add missing details
assert result is not None assert result is not None
if 'type' in result.geometry:
result.geometry['type'] = GEOMETRY_TYPE_MAP.get(result.geometry['type'],
result.geometry['type'])
indexed_date = getattr(row, 'indexed_date', None) indexed_date = getattr(row, 'indexed_date', None)
if indexed_date is not None: if indexed_date is not None:
result.indexed_date = indexed_date.replace(tzinfo=dt.timezone.utc) result.indexed_date = indexed_date.replace(tzinfo=dt.timezone.utc)
@@ -236,3 +238,14 @@ async def get_simple_place(conn: SearchConnection, place: ntyp.PlaceRef,
await nres.add_result_details(conn, [result], details) await nres.add_result_details(conn, [result], details)
return result return result
GEOMETRY_TYPE_MAP = {
'POINT': 'ST_Point',
'MULTIPOINT': 'ST_MultiPoint',
'LINESTRING': 'ST_LineString',
'MULTILINESTRING': 'ST_MultiLineString',
'POLYGON': 'ST_Polygon',
'MULTIPOLYGON': 'ST_MultiPolygon',
'GEOMETRYCOLLECTION': 'ST_GeometryCollection'
}

22
nominatim/api/results.py
View File

@@ -19,7 +19,7 @@ import datetime as dt
import sqlalchemy as sa import sqlalchemy as sa
from nominatim.typing import SaSelect, SaRow from nominatim.typing import SaSelect, SaRow
from nominatim.db.sqlalchemy_functions import CrosscheckNames from nominatim.db.sqlalchemy_types import Geometry
from nominatim.api.types import Point, Bbox, LookupDetails from nominatim.api.types import Point, Bbox, LookupDetails
from nominatim.api.connection import SearchConnection from nominatim.api.connection import SearchConnection
from nominatim.api.logging import log from nominatim.api.logging import log
@@ -589,7 +589,7 @@ async def complete_address_details(conn: SearchConnection, results: List[BaseRes
if not lookup_ids: if not lookup_ids:
return return
ltab = sa.func.json_array_elements(sa.type_coerce(lookup_ids, sa.JSON))\ ltab = sa.func.JsonArrayEach(sa.type_coerce(lookup_ids, sa.JSON))\
.table_valued(sa.column('value', type_=sa.JSON)) # type: ignore[no-untyped-call] .table_valued(sa.column('value', type_=sa.JSON)) # type: ignore[no-untyped-call]
t = conn.t.placex t = conn.t.placex
@@ -608,7 +608,7 @@ async def complete_address_details(conn: SearchConnection, results: List[BaseRes
.order_by('src_place_id')\ .order_by('src_place_id')\
.order_by(sa.column('rank_address').desc())\ .order_by(sa.column('rank_address').desc())\
.order_by((taddr.c.place_id == ltab.c.value['pid'].as_integer()).desc())\ .order_by((taddr.c.place_id == ltab.c.value['pid'].as_integer()).desc())\
.order_by(sa.case((CrosscheckNames(t.c.name, ltab.c.value['names']), 2), .order_by(sa.case((sa.func.CrosscheckNames(t.c.name, ltab.c.value['names']), 2),
(taddr.c.isaddress, 0), (taddr.c.isaddress, 0),
(sa.and_(taddr.c.fromarea, (sa.and_(taddr.c.fromarea,
t.c.geometry.ST_Contains( t.c.geometry.ST_Contains(
@@ -652,7 +652,7 @@ async def complete_address_details(conn: SearchConnection, results: List[BaseRes
parent_lookup_ids = list(filter(lambda e: e['pid'] != e['lid'], lookup_ids)) parent_lookup_ids = list(filter(lambda e: e['pid'] != e['lid'], lookup_ids))
if parent_lookup_ids: if parent_lookup_ids:
ltab = sa.func.json_array_elements(sa.type_coerce(parent_lookup_ids, sa.JSON))\ ltab = sa.func.JsonArrayEach(sa.type_coerce(parent_lookup_ids, sa.JSON))\
.table_valued(sa.column('value', type_=sa.JSON)) # type: ignore[no-untyped-call] .table_valued(sa.column('value', type_=sa.JSON)) # type: ignore[no-untyped-call]
sql = sa.select(ltab.c.value['pid'].as_integer().label('src_place_id'), sql = sa.select(ltab.c.value['pid'].as_integer().label('src_place_id'),
t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name, t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
@@ -687,14 +687,10 @@ def _placex_select_address_row(conn: SearchConnection,
return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name, return sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
t.c.class_.label('class'), t.c.type, t.c.class_.label('class'), t.c.type,
t.c.admin_level, t.c.housenumber, t.c.admin_level, t.c.housenumber,
sa.literal_column("""ST_GeometryType(geometry) in t.c.geometry.is_area().label('fromarea'),
('ST_Polygon','ST_MultiPolygon')""").label('fromarea'),
t.c.rank_address, t.c.rank_address,
sa.literal_column( t.c.geometry.distance_spheroid(
f"""ST_DistanceSpheroid(geometry, sa.bindparam('centroid', value=centroid, type_=Geometry)).label('distance'))
'SRID=4326;{centroid.to_wkt()}'::geometry,
'SPHEROID["WGS 84",6378137,298.257223563, AUTHORITY["EPSG","7030"]]')
""").label('distance'))
async def complete_linked_places(conn: SearchConnection, result: BaseResult) -> None: async def complete_linked_places(conn: SearchConnection, result: BaseResult) -> None:
@@ -728,10 +724,10 @@ async def complete_keywords(conn: SearchConnection, result: BaseResult) -> None:
sel = sa.select(t.c.word_id, t.c.word_token, t.c.word) sel = sa.select(t.c.word_id, t.c.word_token, t.c.word)
for name_tokens, address_tokens in await conn.execute(sql): for name_tokens, address_tokens in await conn.execute(sql):
for row in await conn.execute(sel.where(t.c.word_id == sa.any_(name_tokens))): for row in await conn.execute(sel.where(t.c.word_id.in_(name_tokens))):
result.name_keywords.append(WordInfo(*row)) result.name_keywords.append(WordInfo(*row))
for row in await conn.execute(sel.where(t.c.word_id == sa.any_(address_tokens))): for row in await conn.execute(sel.where(t.c.word_id.in_(address_tokens))):
result.address_keywords.append(WordInfo(*row)) result.address_keywords.append(WordInfo(*row))

43
nominatim/db/sqlalchemy_functions.py
View File

@@ -10,7 +10,6 @@ Custom functions and expressions for SQLAlchemy.
from typing import Any from typing import Any
import sqlalchemy as sa import sqlalchemy as sa
from sqlalchemy.sql.expression import FunctionElement
from sqlalchemy.ext.compiler import compiles from sqlalchemy.ext.compiler import compiles
from nominatim.typing import SaColumn from nominatim.typing import SaColumn
@@ -41,10 +40,11 @@ def select_index_placex_geometry_reverse_lookupplacenode(table: str) -> 'sa.Text
f" AND {table}.osm_type = 'N'") f" AND {table}.osm_type = 'N'")
class CrosscheckNames(FunctionElement[Any]): class CrosscheckNames(sa.sql.functions.GenericFunction[bool]):
""" Check if in the given list of names in parameters 1 any of the names """ Check if in the given list of names in parameters 1 any of the names
from the JSON array in parameter 2 are contained. from the JSON array in parameter 2 are contained.
""" """
type = sa.Boolean()
name = 'CrosscheckNames' name = 'CrosscheckNames'
inherit_cache = True inherit_cache = True
@@ -54,3 +54,42 @@ def compile_crosscheck_names(element: SaColumn,
arg1, arg2 = list(element.clauses) arg1, arg2 = list(element.clauses)
return "coalesce(avals(%s) && ARRAY(SELECT * FROM json_array_elements_text(%s)), false)" % ( return "coalesce(avals(%s) && ARRAY(SELECT * FROM json_array_elements_text(%s)), false)" % (
compiler.process(arg1, **kw), compiler.process(arg2, **kw)) compiler.process(arg1, **kw), compiler.process(arg2, **kw))
@compiles(CrosscheckNames, 'sqlite') # type: ignore[no-untyped-call, misc]
def compile_sqlite_crosscheck_names(element: SaColumn,
compiler: 'sa.Compiled', **kw: Any) -> str:
arg1, arg2 = list(element.clauses)
return "EXISTS(SELECT *"\
" FROM json_each(%s) as name, json_each(%s) as match_name"\
" WHERE name.value = match_name.value)"\
% (compiler.process(arg1, **kw), compiler.process(arg2, **kw))
class JsonArrayEach(sa.sql.functions.GenericFunction[Any]):
""" Return elements of a json array as a set.
"""
name = 'JsonArrayEach'
inherit_cache = True
@compiles(JsonArrayEach) # type: ignore[no-untyped-call, misc]
def default_json_array_each(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
return "json_array_elements(%s)" % compiler.process(element.clauses, **kw)
@compiles(JsonArrayEach, 'sqlite') # type: ignore[no-untyped-call, misc]
def sqlite_json_array_each(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
return "json_each(%s)" % compiler.process(element.clauses, **kw)
class Greatest(sa.sql.functions.GenericFunction[Any]):
""" Function to compute maximum of all its input parameters.
"""
name = 'greatest'
inherit_cache = True
@compiles(Greatest, 'sqlite') # type: ignore[no-untyped-call, misc]
def sqlite_greatest(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
return "max(%s)" % compiler.process(element.clauses, **kw)

68
nominatim/db/sqlalchemy_types.py
View File

@@ -18,29 +18,26 @@ from nominatim.typing import SaColumn, SaBind
#pylint: disable=all #pylint: disable=all
SQLITE_FUNCTION_ALIAS = ( class Geometry_DistanceSpheroid(sa.sql.expression.FunctionElement[float]):
('ST_AsEWKB', sa.Text, 'AsEWKB'), """ Function to compute the spherical distance in meters.
('ST_AsGeoJSON', sa.Text, 'AsGeoJSON'), """
('ST_AsKML', sa.Text, 'AsKML'), type = sa.Float()
('ST_AsSVG', sa.Text, 'AsSVG'), name = 'Geometry_DistanceSpheroid'
) inherit_cache = True
def _add_function_alias(func: str, ftype: type, alias: str) -> None:
_FuncDef = type(func, (sa.sql.functions.GenericFunction, ), {
"type": ftype,
"name": func,
"identifier": func,
"inherit_cache": True})
func_templ = f"{alias}(%s)" @compiles(Geometry_DistanceSpheroid) # type: ignore[no-untyped-call, misc]
def _default_distance_spheroid(element: SaColumn,
compiler: 'sa.Compiled', **kw: Any) -> str:
return "ST_DistanceSpheroid(%s,"\
" 'SPHEROID[\"WGS 84\",6378137,298.257223563, AUTHORITY[\"EPSG\",\"7030\"]]')"\
% compiler.process(element.clauses, **kw)
def _sqlite_impl(element: Any, compiler: Any, **kw: Any) -> Any:
return func_templ % compiler.process(element.clauses, **kw)
compiles(_FuncDef, 'sqlite')(_sqlite_impl) # type: ignore[no-untyped-call] @compiles(Geometry_DistanceSpheroid, 'sqlite') # type: ignore[no-untyped-call, misc]
def _spatialite_distance_spheroid(element: SaColumn,
for alias in SQLITE_FUNCTION_ALIAS: compiler: 'sa.Compiled', **kw: Any) -> str:
_add_function_alias(*alias) return "Distance(%s, true)" % compiler.process(element.clauses, **kw)
class Geometry(types.UserDefinedType): # type: ignore[type-arg] class Geometry(types.UserDefinedType): # type: ignore[type-arg]
@@ -148,6 +145,39 @@ class Geometry(types.UserDefinedType): # type: ignore[type-arg]
return sa.func.ST_LineLocatePoint(self, other, type_=sa.Float) return sa.func.ST_LineLocatePoint(self, other, type_=sa.Float)
def distance_spheroid(self, other: SaColumn) -> SaColumn:
return Geometry_DistanceSpheroid(self, other)
@compiles(Geometry, 'sqlite') # type: ignore[no-untyped-call] @compiles(Geometry, 'sqlite') # type: ignore[no-untyped-call]
def get_col_spec(self, *args, **kwargs): # type: ignore[no-untyped-def] def get_col_spec(self, *args, **kwargs): # type: ignore[no-untyped-def]
return 'GEOMETRY' return 'GEOMETRY'
SQLITE_FUNCTION_ALIAS = (
('ST_AsEWKB', sa.Text, 'AsEWKB'),
('ST_GeomFromEWKT', Geometry, 'GeomFromEWKT'),
('ST_AsGeoJSON', sa.Text, 'AsGeoJSON'),
('ST_AsKML', sa.Text, 'AsKML'),
('ST_AsSVG', sa.Text, 'AsSVG'),
)
def _add_function_alias(func: str, ftype: type, alias: str) -> None:
_FuncDef = type(func, (sa.sql.functions.GenericFunction, ), {
"type": ftype,
"name": func,
"identifier": func,
"inherit_cache": True})
func_templ = f"{alias}(%s)"
def _sqlite_impl(element: Any, compiler: Any, **kw: Any) -> Any:
return func_templ % compiler.process(element.clauses, **kw)
compiles(_FuncDef, 'sqlite')(_sqlite_impl) # type: ignore[no-untyped-call]
for alias in SQLITE_FUNCTION_ALIAS:
_add_function_alias(*alias)

89
test/python/api/test_api_details.py
View File

@@ -15,7 +15,7 @@ import nominatim.api as napi
@pytest.mark.parametrize('idobj', (napi.PlaceID(332), napi.OsmID('W', 4), @pytest.mark.parametrize('idobj', (napi.PlaceID(332), napi.OsmID('W', 4),
napi.OsmID('W', 4, 'highway'))) napi.OsmID('W', 4, 'highway')))
def test_lookup_in_placex(apiobj, idobj): def test_lookup_in_placex(apiobj, frontend, idobj):
import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0) import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4, apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
class_='highway', type='residential', class_='highway', type='residential',
@@ -31,7 +31,8 @@ def test_lookup_in_placex(apiobj, idobj):
indexed_date=import_date, indexed_date=import_date,
geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)') geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
result = apiobj.api.details(idobj) api = frontend(apiobj, options={'details'})
result = api.details(idobj)
assert result is not None assert result is not None
@@ -69,7 +70,7 @@ def test_lookup_in_placex(apiobj, idobj):
assert result.geometry == {'type': 'ST_LineString'} assert result.geometry == {'type': 'ST_LineString'}
def test_lookup_in_placex_minimal_info(apiobj): def test_lookup_in_placex_minimal_info(apiobj, frontend):
import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0) import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4, apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
class_='highway', type='residential', class_='highway', type='residential',
@@ -79,7 +80,8 @@ def test_lookup_in_placex_minimal_info(apiobj):
indexed_date=import_date, indexed_date=import_date,
geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)') geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
result = apiobj.api.details(napi.PlaceID(332)) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332))
assert result is not None assert result is not None
@@ -117,16 +119,17 @@ def test_lookup_in_placex_minimal_info(apiobj):
assert result.geometry == {'type': 'ST_LineString'} assert result.geometry == {'type': 'ST_LineString'}
def test_lookup_in_placex_with_geometry(apiobj): def test_lookup_in_placex_with_geometry(apiobj, frontend):
apiobj.add_placex(place_id=332, apiobj.add_placex(place_id=332,
geometry='LINESTRING(23 34, 23.1 34)') geometry='LINESTRING(23 34, 23.1 34)')
result = apiobj.api.details(napi.PlaceID(332), geometry_output=napi.GeometryFormat.GEOJSON) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332), geometry_output=napi.GeometryFormat.GEOJSON)
assert result.geometry == {'geojson': '{"type":"LineString","coordinates":[[23,34],[23.1,34]]}'} assert result.geometry == {'geojson': '{"type":"LineString","coordinates":[[23,34],[23.1,34]]}'}
def test_lookup_placex_with_address_details(apiobj): def test_lookup_placex_with_address_details(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4, apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
class_='highway', type='residential', name='Street', class_='highway', type='residential', name='Street',
country_code='pl', country_code='pl',
@@ -143,7 +146,8 @@ def test_lookup_placex_with_address_details(apiobj):
country_code='pl', country_code='pl',
rank_search=17, rank_address=16) rank_search=17, rank_address=16)
result = apiobj.api.details(napi.PlaceID(332), address_details=True) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332), address_details=True)
assert result.address_rows == [ assert result.address_rows == [
napi.AddressLine(place_id=332, osm_object=('W', 4), napi.AddressLine(place_id=332, osm_object=('W', 4),
@@ -172,18 +176,19 @@ def test_lookup_placex_with_address_details(apiobj):
] ]
def test_lookup_place_with_linked_places_none_existing(apiobj): def test_lookup_place_with_linked_places_none_existing(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4, apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
class_='highway', type='residential', name='Street', class_='highway', type='residential', name='Street',
country_code='pl', linked_place_id=45, country_code='pl', linked_place_id=45,
rank_search=27, rank_address=26) rank_search=27, rank_address=26)
result = apiobj.api.details(napi.PlaceID(332), linked_places=True) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332), linked_places=True)
assert result.linked_rows == [] assert result.linked_rows == []
def test_lookup_place_with_linked_places_existing(apiobj): def test_lookup_place_with_linked_places_existing(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4, apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
class_='highway', type='residential', name='Street', class_='highway', type='residential', name='Street',
country_code='pl', linked_place_id=45, country_code='pl', linked_place_id=45,
@@ -197,7 +202,8 @@ def test_lookup_place_with_linked_places_existing(apiobj):
country_code='pl', linked_place_id=332, country_code='pl', linked_place_id=332,
rank_search=27, rank_address=26) rank_search=27, rank_address=26)
result = apiobj.api.details(napi.PlaceID(332), linked_places=True) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332), linked_places=True)
assert result.linked_rows == [ assert result.linked_rows == [
napi.AddressLine(place_id=1001, osm_object=('W', 5), napi.AddressLine(place_id=1001, osm_object=('W', 5),
@@ -213,18 +219,19 @@ def test_lookup_place_with_linked_places_existing(apiobj):
] ]
def test_lookup_place_with_parented_places_not_existing(apiobj): def test_lookup_place_with_parented_places_not_existing(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4, apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
class_='highway', type='residential', name='Street', class_='highway', type='residential', name='Street',
country_code='pl', parent_place_id=45, country_code='pl', parent_place_id=45,
rank_search=27, rank_address=26) rank_search=27, rank_address=26)
result = apiobj.api.details(napi.PlaceID(332), parented_places=True) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332), parented_places=True)
assert result.parented_rows == [] assert result.parented_rows == []
def test_lookup_place_with_parented_places_existing(apiobj): def test_lookup_place_with_parented_places_existing(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4, apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
class_='highway', type='residential', name='Street', class_='highway', type='residential', name='Street',
country_code='pl', parent_place_id=45, country_code='pl', parent_place_id=45,
@@ -238,7 +245,8 @@ def test_lookup_place_with_parented_places_existing(apiobj):
country_code='pl', parent_place_id=332, country_code='pl', parent_place_id=332,
rank_search=27, rank_address=26) rank_search=27, rank_address=26)
result = apiobj.api.details(napi.PlaceID(332), parented_places=True) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332), parented_places=True)
assert result.parented_rows == [ assert result.parented_rows == [
napi.AddressLine(place_id=1001, osm_object=('N', 5), napi.AddressLine(place_id=1001, osm_object=('N', 5),
@@ -250,7 +258,7 @@ def test_lookup_place_with_parented_places_existing(apiobj):
@pytest.mark.parametrize('idobj', (napi.PlaceID(4924), napi.OsmID('W', 9928))) @pytest.mark.parametrize('idobj', (napi.PlaceID(4924), napi.OsmID('W', 9928)))
def test_lookup_in_osmline(apiobj, idobj): def test_lookup_in_osmline(apiobj, frontend, idobj):
import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0) import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
apiobj.add_osmline(place_id=4924, osm_id=9928, apiobj.add_osmline(place_id=4924, osm_id=9928,
parent_place_id=12, parent_place_id=12,
@@ -260,7 +268,8 @@ def test_lookup_in_osmline(apiobj, idobj):
indexed_date=import_date, indexed_date=import_date,
geometry='LINESTRING(23 34, 23 35)') geometry='LINESTRING(23 34, 23 35)')
result = apiobj.api.details(idobj) api = frontend(apiobj, options={'details'})
result = api.details(idobj)
assert result is not None assert result is not None
@@ -298,7 +307,7 @@ def test_lookup_in_osmline(apiobj, idobj):
assert result.geometry == {'type': 'ST_LineString'} assert result.geometry == {'type': 'ST_LineString'}
def test_lookup_in_osmline_split_interpolation(apiobj): def test_lookup_in_osmline_split_interpolation(apiobj, frontend):
apiobj.add_osmline(place_id=1000, osm_id=9, apiobj.add_osmline(place_id=1000, osm_id=9,
startnumber=2, endnumber=4, step=1) startnumber=2, endnumber=4, step=1)
apiobj.add_osmline(place_id=1001, osm_id=9, apiobj.add_osmline(place_id=1001, osm_id=9,
@@ -306,18 +315,19 @@ def test_lookup_in_osmline_split_interpolation(apiobj):
apiobj.add_osmline(place_id=1002, osm_id=9, apiobj.add_osmline(place_id=1002, osm_id=9,
startnumber=11, endnumber=20, step=1) startnumber=11, endnumber=20, step=1)
api = frontend(apiobj, options={'details'})
for i in range(1, 6): for i in range(1, 6):
result = apiobj.api.details(napi.OsmID('W', 9, str(i))) result = api.details(napi.OsmID('W', 9, str(i)))
assert result.place_id == 1000 assert result.place_id == 1000
for i in range(7, 11): for i in range(7, 11):
result = apiobj.api.details(napi.OsmID('W', 9, str(i))) result = api.details(napi.OsmID('W', 9, str(i)))
assert result.place_id == 1001 assert result.place_id == 1001
for i in range(12, 22): for i in range(12, 22):
result = apiobj.api.details(napi.OsmID('W', 9, str(i))) result = api.details(napi.OsmID('W', 9, str(i)))
assert result.place_id == 1002 assert result.place_id == 1002
def test_lookup_osmline_with_address_details(apiobj): def test_lookup_osmline_with_address_details(apiobj, frontend):
apiobj.add_osmline(place_id=9000, osm_id=9, apiobj.add_osmline(place_id=9000, osm_id=9,
startnumber=2, endnumber=4, step=1, startnumber=2, endnumber=4, step=1,
parent_place_id=332) parent_place_id=332)
@@ -337,7 +347,8 @@ def test_lookup_osmline_with_address_details(apiobj):
country_code='pl', country_code='pl',
rank_search=17, rank_address=16) rank_search=17, rank_address=16)
result = apiobj.api.details(napi.PlaceID(9000), address_details=True) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(9000), address_details=True)
assert result.address_rows == [ assert result.address_rows == [
napi.AddressLine(place_id=332, osm_object=('W', 4), napi.AddressLine(place_id=332, osm_object=('W', 4),
@@ -366,7 +377,7 @@ def test_lookup_osmline_with_address_details(apiobj):
] ]
def test_lookup_in_tiger(apiobj): def test_lookup_in_tiger(apiobj, frontend):
apiobj.add_tiger(place_id=4924, apiobj.add_tiger(place_id=4924,
parent_place_id=12, parent_place_id=12,
startnumber=1, endnumber=4, step=1, startnumber=1, endnumber=4, step=1,
@@ -377,7 +388,8 @@ def test_lookup_in_tiger(apiobj):
osm_type='W', osm_id=6601223, osm_type='W', osm_id=6601223,
geometry='LINESTRING(23 34, 23 35)') geometry='LINESTRING(23 34, 23 35)')
result = apiobj.api.details(napi.PlaceID(4924)) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(4924))
assert result is not None assert result is not None
@@ -415,7 +427,7 @@ def test_lookup_in_tiger(apiobj):
assert result.geometry == {'type': 'ST_LineString'} assert result.geometry == {'type': 'ST_LineString'}
def test_lookup_tiger_with_address_details(apiobj): def test_lookup_tiger_with_address_details(apiobj, frontend):
apiobj.add_tiger(place_id=9000, apiobj.add_tiger(place_id=9000,
startnumber=2, endnumber=4, step=1, startnumber=2, endnumber=4, step=1,
parent_place_id=332) parent_place_id=332)
@@ -435,7 +447,8 @@ def test_lookup_tiger_with_address_details(apiobj):
country_code='us', country_code='us',
rank_search=17, rank_address=16) rank_search=17, rank_address=16)
result = apiobj.api.details(napi.PlaceID(9000), address_details=True) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(9000), address_details=True)
assert result.address_rows == [ assert result.address_rows == [
napi.AddressLine(place_id=332, osm_object=('W', 4), napi.AddressLine(place_id=332, osm_object=('W', 4),
@@ -464,7 +477,7 @@ def test_lookup_tiger_with_address_details(apiobj):
] ]
def test_lookup_in_postcode(apiobj): def test_lookup_in_postcode(apiobj, frontend):
import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0) import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
apiobj.add_postcode(place_id=554, apiobj.add_postcode(place_id=554,
parent_place_id=152, parent_place_id=152,
@@ -474,7 +487,8 @@ def test_lookup_in_postcode(apiobj):
indexed_date=import_date, indexed_date=import_date,
geometry='POINT(-9.45 5.6)') geometry='POINT(-9.45 5.6)')
result = apiobj.api.details(napi.PlaceID(554)) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(554))
assert result is not None assert result is not None
@@ -512,7 +526,7 @@ def test_lookup_in_postcode(apiobj):
assert result.geometry == {'type': 'ST_Point'} assert result.geometry == {'type': 'ST_Point'}
def test_lookup_postcode_with_address_details(apiobj): def test_lookup_postcode_with_address_details(apiobj, frontend):
apiobj.add_postcode(place_id=9000, apiobj.add_postcode(place_id=9000,
parent_place_id=332, parent_place_id=332,
postcode='34 425', postcode='34 425',
@@ -528,7 +542,8 @@ def test_lookup_postcode_with_address_details(apiobj):
country_code='gb', country_code='gb',
rank_search=17, rank_address=16) rank_search=17, rank_address=16)
result = apiobj.api.details(napi.PlaceID(9000), address_details=True) api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(9000), address_details=True)
assert result.address_rows == [ assert result.address_rows == [
napi.AddressLine(place_id=9000, osm_object=None, napi.AddressLine(place_id=9000, osm_object=None,
@@ -559,18 +574,20 @@ def test_lookup_postcode_with_address_details(apiobj):
@pytest.mark.parametrize('objid', [napi.PlaceID(1736), @pytest.mark.parametrize('objid', [napi.PlaceID(1736),
napi.OsmID('W', 55), napi.OsmID('W', 55),
napi.OsmID('N', 55, 'amenity')]) napi.OsmID('N', 55, 'amenity')])
def test_lookup_missing_object(apiobj, objid): def test_lookup_missing_object(apiobj, frontend, objid):
apiobj.add_placex(place_id=1, osm_type='N', osm_id=55, apiobj.add_placex(place_id=1, osm_type='N', osm_id=55,
class_='place', type='suburb') class_='place', type='suburb')
assert apiobj.api.details(objid) is None api = frontend(apiobj, options={'details'})
assert api.details(objid) is None
@pytest.mark.parametrize('gtype', (napi.GeometryFormat.KML, @pytest.mark.parametrize('gtype', (napi.GeometryFormat.KML,
napi.GeometryFormat.SVG, napi.GeometryFormat.SVG,
napi.GeometryFormat.TEXT)) napi.GeometryFormat.TEXT))
def test_lookup_unsupported_geometry(apiobj, gtype): def test_lookup_unsupported_geometry(apiobj, frontend, gtype):
apiobj.add_placex(place_id=332) apiobj.add_placex(place_id=332)
api = frontend(apiobj, options={'details'})
with pytest.raises(ValueError): with pytest.raises(ValueError):
apiobj.api.details(napi.PlaceID(332), geometry_output=gtype) api.details(napi.PlaceID(332), geometry_output=gtype)