uses: shivammathur/setup-php@v2
with:
php-version: ${{ matrix.php }}
- tools: phpunit, phpcs, composer
+ tools: phpunit:9, phpcs, composer
ini-values: opcache.jit=disable
- uses: actions/setup-python@v4
# 'too-many-ancestors' is triggered already by deriving from UserDict
# 'not-context-manager' disabled because it causes false positives once
# typed Python is enabled. See also https://github.com/PyCQA/pylint/issues/5273
-disable=too-few-public-methods,duplicate-code,too-many-ancestors,bad-option-value,no-self-use,not-context-manager
+disable=too-few-public-methods,duplicate-code,too-many-ancestors,bad-option-value,no-self-use,not-context-manager,use-dict-literal
-good-names=i,x,y,m,fd,db,cc
+good-names=i,x,y,m,t,fd,db,cc
# See also https://github.com/PyCQA/pylint/issues/6006
# pylint: disable=useless-import-alias
-from nominatim.api.core import (NominatimAPI as NominatimAPI,
- NominatimAPIAsync as NominatimAPIAsync)
-from nominatim.api.status import (StatusResult as StatusResult)
+from .core import (NominatimAPI as NominatimAPI,
+ NominatimAPIAsync as NominatimAPIAsync)
+from .status import (StatusResult as StatusResult)
+from .types import (PlaceID as PlaceID,
+ OsmID as OsmID,
+ PlaceRef as PlaceRef,
+ Point as Point,
+ GeometryFormat as GeometryFormat,
+ LookupDetails as LookupDetails)
+from .results import (SourceTable as SourceTable,
+ AddressLine as AddressLine,
+ AddressLines as AddressLines,
+ WordInfo as WordInfo,
+ WordInfos as WordInfos,
+ SearchResult as SearchResult)
+from .localization import (Locales as Locales)
from nominatim.db.sqlalchemy_schema import SearchTables
from nominatim.config import Configuration
-from nominatim.api.status import get_status, StatusResult
from nominatim.api.connection import SearchConnection
+from nominatim.api.status import get_status, StatusResult
+from nominatim.api.lookup import get_place_by_id
+from nominatim.api.types import PlaceRef, LookupDetails
+from nominatim.api.results import SearchResult
+
class NominatimAPIAsync:
""" API loader asynchornous version.
return status
+ async def lookup(self, place: PlaceRef,
+ details: LookupDetails) -> Optional[SearchResult]:
+ """ Get detailed information about a place in the database.
+
+ Returns None if there is no entry under the given ID.
+ """
+ async with self.begin() as db:
+ return await get_place_by_id(db, place, details)
+
+
class NominatimAPI:
""" API loader, synchronous version.
"""
self._loop.close()
+ @property
+ def config(self) -> Configuration:
+ """ Return the configuration used by the API.
+ """
+ return self._async_api.config
+
def status(self) -> StatusResult:
""" Return the status of the database.
"""
return self._loop.run_until_complete(self._async_api.status())
+
+
+ def lookup(self, place: PlaceRef,
+ details: LookupDetails) -> Optional[SearchResult]:
+ """ Get detailed information about a place in the database.
+ """
+ return self._loop.run_until_complete(self._async_api.lookup(place, details))
--- /dev/null
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# 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.
+"""
+Helper functions for localizing names of results.
+"""
+from typing import Mapping, List, Optional
+
+import re
+
+class Locales:
+ """ Helper class for localization of names.
+
+ It takes a list of language prefixes in their order of preferred
+ usage.
+ """
+
+ def __init__(self, langs: Optional[List[str]] = None):
+ self.languages = langs or []
+ self.name_tags: List[str] = []
+
+ # Build the list of supported tags. It is currently hard-coded.
+ self._add_lang_tags('name')
+ self._add_tags('name', 'brand')
+ self._add_lang_tags('official_name', 'short_name')
+ self._add_tags('official_name', 'short_name', 'ref')
+
+
+ def __bool__(self) -> bool:
+ return len(self.languages) > 0
+
+
+ def _add_tags(self, *tags: str) -> None:
+ for tag in tags:
+ self.name_tags.append(tag)
+ self.name_tags.append(f"_place_{tag}")
+
+
+ def _add_lang_tags(self, *tags: str) -> None:
+ for tag in tags:
+ for lang in self.languages:
+ self.name_tags.append(f"{tag}:{lang}")
+ self.name_tags.append(f"_place_{tag}:{lang}")
+
+
+ def display_name(self, names: Optional[Mapping[str, str]]) -> str:
+ """ Return the best matching name from a dictionary of names
+ containing different name variants.
+
+ If 'names' is null or empty, an empty string is returned. If no
+ appropriate localization is found, the first name is returned.
+ """
+ if not names:
+ return ''
+
+ if len(names) > 1:
+ for tag in self.name_tags:
+ if tag in names:
+ return names[tag]
+
+ # Nothing? Return any of the other names as a default.
+ return next(iter(names.values()))
+
+
+ @staticmethod
+ def from_accept_languages(langstr: str) -> 'Locales':
+ """ Create a localization object from a language list in the
+ format of HTTP accept-languages header.
+
+ The functions tries to be forgiving of format errors by first splitting
+ the string into comma-separated parts and then parsing each
+ description separately. Badly formatted parts are then ignored.
+ """
+ # split string into languages
+ candidates = []
+ for desc in langstr.split(','):
+ m = re.fullmatch(r'\s*([a-z_-]+)(?:;\s*q\s*=\s*([01](?:\.\d+)?))?\s*',
+ desc, flags=re.I)
+ if m:
+ candidates.append((m[1], float(m[2] or 1.0)))
+
+ # sort the results by the weight of each language (preserving order).
+ candidates.sort(reverse=True, key=lambda e: e[1])
+
+ # If a language has a region variant, also add the language without
+ # variant but only if it isn't already in the list to not mess up the weight.
+ languages = []
+ for lid, _ in candidates:
+ languages.append(lid)
+ parts = lid.split('-', 1)
+ if len(parts) > 1 and all(c[0] != parts[0] for c in candidates):
+ languages.append(parts[0])
+
+ return Locales(languages)
--- /dev/null
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# 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.
+"""
+Implementation of place lookup by ID.
+"""
+from typing import Optional
+
+import sqlalchemy as sa
+
+from nominatim.typing import SaColumn, SaLabel, SaRow
+from nominatim.api.connection import SearchConnection
+import nominatim.api.types as ntyp
+import nominatim.api.results as nres
+
+def _select_column_geometry(column: SaColumn,
+ geometry_output: ntyp.GeometryFormat) -> SaLabel:
+ """ Create the appropriate column expression for selecting a
+ geometry for the details response.
+ """
+ if geometry_output & ntyp.GeometryFormat.GEOJSON:
+ return sa.literal_column(f"""
+ ST_AsGeoJSON(CASE WHEN ST_NPoints({column.name}) > 5000
+ THEN ST_SimplifyPreserveTopology({column.name}, 0.0001)
+ ELSE {column.name} END)
+ """).label('geometry_geojson')
+
+ return sa.func.ST_GeometryType(column).label('geometry_type')
+
+
+async def find_in_placex(conn: SearchConnection, place: ntyp.PlaceRef,
+ details: ntyp.LookupDetails) -> Optional[SaRow]:
+ """ Search for the given place in the placex table and return the
+ base information.
+ """
+ t = conn.t.placex
+ sql = sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
+ t.c.class_, t.c.type, t.c.admin_level,
+ t.c.address, t.c.extratags,
+ t.c.housenumber, t.c.postcode, t.c.country_code,
+ t.c.importance, t.c.wikipedia, t.c.indexed_date,
+ t.c.parent_place_id, t.c.rank_address, t.c.rank_search,
+ t.c.linked_place_id,
+ sa.func.ST_X(t.c.centroid).label('x'),
+ sa.func.ST_Y(t.c.centroid).label('y'),
+ _select_column_geometry(t.c.geometry, details.geometry_output))
+
+ if isinstance(place, ntyp.PlaceID):
+ sql = sql.where(t.c.place_id == place.place_id)
+ elif isinstance(place, ntyp.OsmID):
+ sql = sql.where(t.c.osm_type == place.osm_type)\
+ .where(t.c.osm_id == place.osm_id)
+ if place.osm_class:
+ sql = sql.where(t.c.class_ == place.osm_class)
+ else:
+ sql = sql.order_by(t.c.class_)
+ sql = sql.limit(1)
+ else:
+ return None
+
+ return (await conn.execute(sql)).one_or_none()
+
+
+async def find_in_osmline(conn: SearchConnection, place: ntyp.PlaceRef,
+ details: ntyp.LookupDetails) -> Optional[SaRow]:
+ """ Search for the given place in the osmline table and return the
+ base information.
+ """
+ t = conn.t.osmline
+ sql = sa.select(t.c.place_id, t.c.osm_id, t.c.parent_place_id,
+ t.c.indexed_date, t.c.startnumber, t.c.endnumber,
+ t.c.step, t.c.address, t.c.postcode, t.c.country_code,
+ sa.func.ST_X(sa.func.ST_Centroid(t.c.linegeo)).label('x'),
+ sa.func.ST_Y(sa.func.ST_Centroid(t.c.linegeo)).label('y'),
+ _select_column_geometry(t.c.linegeo, details.geometry_output))
+
+ if isinstance(place, ntyp.PlaceID):
+ sql = sql.where(t.c.place_id == place.place_id)
+ elif isinstance(place, ntyp.OsmID) and place.osm_type == 'W':
+ # There may be multiple interpolations for a single way.
+ # If 'class' contains a number, return the one that belongs to that number.
+ sql = sql.where(t.c.osm_id == place.osm_id).limit(1)
+ if place.osm_class and place.osm_class.isdigit():
+ sql = sql.order_by(sa.func.greatest(0,
+ sa.func.least(int(place.osm_class) - t.c.endnumber),
+ t.c.startnumber - int(place.osm_class)))
+ else:
+ return None
+
+ return (await conn.execute(sql)).one_or_none()
+
+
+async def find_in_tiger(conn: SearchConnection, place: ntyp.PlaceRef,
+ details: ntyp.LookupDetails) -> Optional[SaRow]:
+ """ Search for the given place in the table of Tiger addresses and return
+ the base information. Only lookup by place ID is supported.
+ """
+ t = conn.t.tiger
+ sql = sa.select(t.c.place_id, t.c.parent_place_id,
+ t.c.startnumber, t.c.endnumber, t.c.step,
+ t.c.postcode,
+ sa.func.ST_X(sa.func.ST_Centroid(t.c.linegeo)).label('x'),
+ sa.func.ST_Y(sa.func.ST_Centroid(t.c.linegeo)).label('y'),
+ _select_column_geometry(t.c.linegeo, details.geometry_output))
+
+ if isinstance(place, ntyp.PlaceID):
+ sql = sql.where(t.c.place_id == place.place_id)
+ else:
+ return None
+
+ return (await conn.execute(sql)).one_or_none()
+
+
+async def find_in_postcode(conn: SearchConnection, place: ntyp.PlaceRef,
+ details: ntyp.LookupDetails) -> Optional[SaRow]:
+ """ Search for the given place in the postcode table and return the
+ base information. Only lookup by place ID is supported.
+ """
+ t = conn.t.postcode
+ sql = sa.select(t.c.place_id, t.c.parent_place_id,
+ t.c.rank_search, t.c.rank_address,
+ t.c.indexed_date, t.c.postcode, t.c.country_code,
+ sa.func.ST_X(t.c.geometry).label('x'),
+ sa.func.ST_Y(t.c.geometry).label('y'),
+ _select_column_geometry(t.c.geometry, details.geometry_output))
+
+ if isinstance(place, ntyp.PlaceID):
+ sql = sql.where(t.c.place_id == place.place_id)
+ else:
+ return None
+
+ return (await conn.execute(sql)).one_or_none()
+
+
+async def get_place_by_id(conn: SearchConnection, place: ntyp.PlaceRef,
+ details: ntyp.LookupDetails) -> Optional[nres.SearchResult]:
+ """ Retrieve a place with additional details from the database.
+ """
+ if details.geometry_output and details.geometry_output != ntyp.GeometryFormat.GEOJSON:
+ raise ValueError("lookup only supports geojosn polygon output.")
+
+ row = await find_in_placex(conn, place, details)
+ if row is not None:
+ result = nres.create_from_placex_row(row)
+ await nres.add_result_details(conn, result, details)
+ return result
+
+ row = await find_in_osmline(conn, place, details)
+ if row is not None:
+ result = nres.create_from_osmline_row(row)
+ await nres.add_result_details(conn, result, details)
+ return result
+
+ row = await find_in_postcode(conn, place, details)
+ if row is not None:
+ result = nres.create_from_postcode_row(row)
+ await nres.add_result_details(conn, result, details)
+ return result
+
+ row = await find_in_tiger(conn, place, details)
+ if row is not None:
+ result = nres.create_from_tiger_row(row)
+ await nres.add_result_details(conn, result, details)
+ return result
+
+ # Nothing found under this ID.
+ return None
"""
Helper classes and functions for formating results into API responses.
"""
-from typing import Type, TypeVar, Dict, List, Callable, Any
+from typing import Type, TypeVar, Dict, List, Callable, Any, Mapping
from collections import defaultdict
T = TypeVar('T') # pylint: disable=invalid-name
-FormatFunc = Callable[[T], str]
+FormatFunc = Callable[[T, Mapping[str, Any]], str]
class FormatDispatcher:
return fmt in self.format_functions[result_type]
- def format_result(self, result: Any, fmt: str) -> str:
+ def format_result(self, result: Any, fmt: str, options: Mapping[str, Any]) -> str:
""" Convert the given result into a string using the given format.
The format is expected to be in the list returned by
`list_formats()`.
"""
- return self.format_functions[type(result)][fmt](result)
+ return self.format_functions[type(result)][fmt](result, options)
--- /dev/null
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# 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.
+"""
+Dataclasses for search results and helper functions to fill them.
+
+Data classes are part of the public API while the functions are for
+internal use only. That's why they are implemented as free-standing functions
+instead of member functions.
+"""
+from typing import Optional, Tuple, Dict, Sequence
+import enum
+import dataclasses
+import datetime as dt
+
+import sqlalchemy as sa
+
+from nominatim.typing import SaSelect, SaRow
+from nominatim.api.types import Point, LookupDetails
+from nominatim.api.connection import SearchConnection
+
+# This file defines complex result data classes.
+# pylint: disable=too-many-instance-attributes
+
+class SourceTable(enum.Enum):
+ """ Enumeration of kinds of results.
+ """
+ PLACEX = 1
+ OSMLINE = 2
+ TIGER = 3
+ POSTCODE = 4
+ COUNTRY = 5
+
+
+@dataclasses.dataclass
+class AddressLine:
+ """ Detailed information about a related place.
+ """
+ place_id: Optional[int]
+ osm_object: Optional[Tuple[str, int]]
+ category: Tuple[str, str]
+ names: Dict[str, str]
+ extratags: Optional[Dict[str, str]]
+
+ admin_level: Optional[int]
+ fromarea: bool
+ isaddress: bool
+ rank_address: int
+ distance: float
+
+
+AddressLines = Sequence[AddressLine]
+
+
+@dataclasses.dataclass
+class WordInfo:
+ """ Detailed information about a search term.
+ """
+ word_id: int
+ word_token: str
+ word: Optional[str] = None
+
+
+WordInfos = Sequence[WordInfo]
+
+
+@dataclasses.dataclass
+class SearchResult:
+ """ Data class collecting all available information about a search result.
+ """
+ source_table: SourceTable
+ category: Tuple[str, str]
+ centroid: Point
+
+ place_id : Optional[int] = None
+ parent_place_id: Optional[int] = None
+ linked_place_id: Optional[int] = None
+ osm_object: Optional[Tuple[str, int]] = None
+ admin_level: int = 15
+
+ names: Optional[Dict[str, str]] = None
+ address: Optional[Dict[str, str]] = None
+ extratags: Optional[Dict[str, str]] = None
+
+ housenumber: Optional[str] = None
+ postcode: Optional[str] = None
+ wikipedia: Optional[str] = None
+
+ rank_address: int = 30
+ rank_search: int = 30
+ importance: Optional[float] = None
+
+ country_code: Optional[str] = None
+
+ indexed_date: Optional[dt.datetime] = None
+
+ address_rows: Optional[AddressLines] = None
+ linked_rows: Optional[AddressLines] = None
+ parented_rows: Optional[AddressLines] = None
+ name_keywords: Optional[WordInfos] = None
+ address_keywords: Optional[WordInfos] = None
+
+ geometry: Dict[str, str] = dataclasses.field(default_factory=dict)
+
+ def __post_init__(self) -> None:
+ if self.indexed_date is not None and self.indexed_date.tzinfo is None:
+ self.indexed_date = self.indexed_date.replace(tzinfo=dt.timezone.utc)
+
+ @property
+ def lat(self) -> float:
+ """ Get the latitude (or y) of the center point of the place.
+ """
+ return self.centroid[1]
+
+
+ @property
+ def lon(self) -> float:
+ """ Get the longitude (or x) of the center point of the place.
+ """
+ return self.centroid[0]
+
+
+ def calculated_importance(self) -> float:
+ """ Get a valid importance value. This is either the stored importance
+ of the value or an artificial value computed from the place's
+ search rank.
+ """
+ return self.importance or (0.7500001 - (self.rank_search/40.0))
+
+
+ # pylint: disable=consider-using-f-string
+ def centroid_as_geojson(self) -> str:
+ """ Get the centroid in GeoJSON format.
+ """
+ return '{"type": "Point","coordinates": [%f, %f]}' % self.centroid
+
+
+def _filter_geometries(row: SaRow) -> Dict[str, str]:
+ return {k[9:]: v for k, v in row._mapping.items() # pylint: disable=W0212
+ if k.startswith('geometry_')}
+
+
+def create_from_placex_row(row: SaRow) -> SearchResult:
+ """ Construct a new SearchResult and add the data from the result row
+ from the placex table.
+ """
+ return SearchResult(source_table=SourceTable.PLACEX,
+ place_id=row.place_id,
+ parent_place_id=row.parent_place_id,
+ linked_place_id=row.linked_place_id,
+ osm_object=(row.osm_type, row.osm_id),
+ category=(row.class_, row.type),
+ admin_level=row.admin_level,
+ names=row.name,
+ address=row.address,
+ extratags=row.extratags,
+ housenumber=row.housenumber,
+ postcode=row.postcode,
+ wikipedia=row.wikipedia,
+ rank_address=row.rank_address,
+ rank_search=row.rank_search,
+ importance=row.importance,
+ country_code=row.country_code,
+ indexed_date=getattr(row, 'indexed_date'),
+ centroid=Point(row.x, row.y),
+ geometry=_filter_geometries(row))
+
+
+def create_from_osmline_row(row: SaRow) -> SearchResult:
+ """ Construct a new SearchResult and add the data from the result row
+ from the osmline table.
+ """
+ return SearchResult(source_table=SourceTable.OSMLINE,
+ place_id=row.place_id,
+ parent_place_id=row.parent_place_id,
+ osm_object=('W', row.osm_id),
+ category=('place', 'houses'),
+ address=row.address,
+ postcode=row.postcode,
+ extratags={'startnumber': str(row.startnumber),
+ 'endnumber': str(row.endnumber),
+ 'step': str(row.step)},
+ country_code=row.country_code,
+ indexed_date=getattr(row, 'indexed_date'),
+ centroid=Point(row.x, row.y),
+ geometry=_filter_geometries(row))
+
+
+def create_from_tiger_row(row: SaRow) -> SearchResult:
+ """ Construct a new SearchResult and add the data from the result row
+ from the Tiger table.
+ """
+ return SearchResult(source_table=SourceTable.TIGER,
+ place_id=row.place_id,
+ parent_place_id=row.parent_place_id,
+ category=('place', 'houses'),
+ postcode=row.postcode,
+ extratags={'startnumber': str(row.startnumber),
+ 'endnumber': str(row.endnumber),
+ 'step': str(row.step)},
+ country_code='us',
+ centroid=Point(row.x, row.y),
+ geometry=_filter_geometries(row))
+
+
+def create_from_postcode_row(row: SaRow) -> SearchResult:
+ """ Construct a new SearchResult and add the data from the result row
+ from the postcode centroid table.
+ """
+ return SearchResult(source_table=SourceTable.POSTCODE,
+ place_id=row.place_id,
+ parent_place_id=row.parent_place_id,
+ category=('place', 'postcode'),
+ names={'ref': row.postcode},
+ rank_search=row.rank_search,
+ rank_address=row.rank_address,
+ country_code=row.country_code,
+ centroid=Point(row.x, row.y),
+ indexed_date=row.indexed_date,
+ geometry=_filter_geometries(row))
+
+
+async def add_result_details(conn: SearchConnection, result: SearchResult,
+ details: LookupDetails) -> None:
+ """ Retrieve more details from the database according to the
+ parameters specified in 'details'.
+ """
+ if details.address_details:
+ await complete_address_details(conn, result)
+ if details.linked_places:
+ await complete_linked_places(conn, result)
+ if details.parented_places:
+ await complete_parented_places(conn, result)
+ if details.keywords:
+ await complete_keywords(conn, result)
+
+
+def _result_row_to_address_row(row: SaRow) -> AddressLine:
+ """ Create a new AddressLine from the results of a datbase query.
+ """
+ extratags: Dict[str, str] = getattr(row, 'extratags', {})
+ if 'place_type' in row:
+ extratags['place_type'] = row.place_type
+
+ names = row.name
+ if getattr(row, 'housenumber', None) is not None:
+ if names is None:
+ names = {}
+ names['housenumber'] = row.housenumber
+
+ return AddressLine(place_id=row.place_id,
+ osm_object=None if row.osm_type is None else (row.osm_type, row.osm_id),
+ category=(getattr(row, 'class'), row.type),
+ names=names,
+ extratags=extratags,
+ admin_level=row.admin_level,
+ fromarea=row.fromarea,
+ isaddress=getattr(row, 'isaddress', True),
+ rank_address=row.rank_address,
+ distance=row.distance)
+
+
+async def complete_address_details(conn: SearchConnection, result: SearchResult) -> None:
+ """ Retrieve information about places that make up the address of the result.
+ """
+ housenumber = -1
+ if result.source_table in (SourceTable.TIGER, SourceTable.OSMLINE):
+ if result.housenumber is not None:
+ housenumber = int(result.housenumber)
+ elif result.extratags is not None and 'startnumber' in result.extratags:
+ # details requests do not come with a specific house number
+ housenumber = int(result.extratags['startnumber'])
+
+ sfn = sa.func.get_addressdata(result.place_id, housenumber)\
+ .table_valued( # type: ignore[no-untyped-call]
+ sa.column('place_id', type_=sa.Integer),
+ 'osm_type',
+ sa.column('osm_id', type_=sa.BigInteger),
+ sa.column('name', type_=conn.t.types.Composite),
+ 'class', 'type', 'place_type',
+ sa.column('admin_level', type_=sa.Integer),
+ sa.column('fromarea', type_=sa.Boolean),
+ sa.column('isaddress', type_=sa.Boolean),
+ sa.column('rank_address', type_=sa.SmallInteger),
+ sa.column('distance', type_=sa.Float))
+ sql = sa.select(sfn).order_by(sa.column('rank_address').desc(),
+ sa.column('isaddress').desc())
+
+ result.address_rows = []
+ for row in await conn.execute(sql):
+ result.address_rows.append(_result_row_to_address_row(row))
+
+# pylint: disable=consider-using-f-string
+def _placex_select_address_row(conn: SearchConnection,
+ centroid: Point) -> SaSelect:
+ t = conn.t.placex
+ 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.admin_level, t.c.housenumber,
+ sa.literal_column("""ST_GeometryType(geometry) in
+ ('ST_Polygon','ST_MultiPolygon')""").label('fromarea'),
+ t.c.rank_address,
+ sa.literal_column(
+ """ST_DistanceSpheroid(geometry, 'SRID=4326;POINT(%f %f)'::geometry,
+ 'SPHEROID["WGS 84",6378137,298.257223563, AUTHORITY["EPSG","7030"]]')
+ """ % centroid).label('distance'))
+
+
+async def complete_linked_places(conn: SearchConnection, result: SearchResult) -> None:
+ """ Retrieve information about places that link to the result.
+ """
+ result.linked_rows = []
+ if result.source_table != SourceTable.PLACEX:
+ return
+
+ sql = _placex_select_address_row(conn, result.centroid)\
+ .where(conn.t.placex.c.linked_place_id == result.place_id)
+
+ for row in await conn.execute(sql):
+ result.linked_rows.append(_result_row_to_address_row(row))
+
+
+async def complete_keywords(conn: SearchConnection, result: SearchResult) -> None:
+ """ Retrieve information about the search terms used for this place.
+ """
+ t = conn.t.search_name
+ sql = sa.select(t.c.name_vector, t.c.nameaddress_vector)\
+ .where(t.c.place_id == result.place_id)
+
+ result.name_keywords = []
+ result.address_keywords = []
+ for name_tokens, address_tokens in await conn.execute(sql):
+ t = conn.t.word
+ sel = sa.select(t.c.word_id, t.c.word_token, t.c.word)
+
+ for row in await conn.execute(sel.where(t.c.word_id == sa.any_(name_tokens))):
+ result.name_keywords.append(WordInfo(*row))
+
+ for row in await conn.execute(sel.where(t.c.word_id == sa.any_(address_tokens))):
+ result.address_keywords.append(WordInfo(*row))
+
+
+async def complete_parented_places(conn: SearchConnection, result: SearchResult) -> None:
+ """ Retrieve information about places that the result provides the
+ address for.
+ """
+ result.parented_rows = []
+ if result.source_table != SourceTable.PLACEX:
+ return
+
+ sql = _placex_select_address_row(conn, result.centroid)\
+ .where(conn.t.placex.c.parent_place_id == result.place_id)\
+ .where(conn.t.placex.c.rank_search == 30)
+
+ for row in await conn.execute(sql):
+ result.parented_rows.append(_result_row_to_address_row(row))
--- /dev/null
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# 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.
+"""
+Complex datatypes used by the Nominatim API.
+"""
+from typing import Optional, Union, NamedTuple
+import dataclasses
+import enum
+
+@dataclasses.dataclass
+class PlaceID:
+ """ Reference an object by Nominatim's internal ID.
+ """
+ place_id: int
+
+
+@dataclasses.dataclass
+class OsmID:
+ """ Reference by the OSM ID and potentially the basic category.
+ """
+ osm_type: str
+ osm_id: int
+ osm_class: Optional[str] = None
+
+ def __post_init__(self) -> None:
+ if self.osm_type not in ('N', 'W', 'R'):
+ raise ValueError(f"Illegal OSM type '{self.osm_type}'. Must be one of N, W, R.")
+
+
+PlaceRef = Union[PlaceID, OsmID]
+
+
+class Point(NamedTuple):
+ """ A geographic point in WGS84 projection.
+ """
+ x: float
+ y: float
+
+
+ @property
+ def lat(self) -> float:
+ """ Return the latitude of the point.
+ """
+ return self.y
+
+
+ @property
+ def lon(self) -> float:
+ """ Return the longitude of the point.
+ """
+ return self.x
+
+
+class GeometryFormat(enum.Flag):
+ """ Geometry output formats supported by Nominatim.
+ """
+ NONE = 0
+ GEOJSON = enum.auto()
+ KML = enum.auto()
+ SVG = enum.auto()
+ TEXT = enum.auto()
+
+
+@dataclasses.dataclass
+class LookupDetails:
+ """ Collection of parameters that define the amount of details
+ returned with a search result.
+ """
+ geometry_output: GeometryFormat = GeometryFormat.NONE
+ """ Add the full geometry of the place to the result. Multiple
+ formats may be selected. Note that geometries can become quite large.
+ """
+ address_details: bool = False
+ """ Get detailed information on the places that make up the address
+ for the result.
+ """
+ linked_places: bool = False
+ """ Get detailed information on the places that link to the result.
+ """
+ parented_places: bool = False
+ """ Get detailed information on all places that this place is a parent
+ for, i.e. all places for which it provides the address details.
+ Only POI places can have parents.
+ """
+ keywords: bool = False
+ """ Add information about the search terms used for this place.
+ """
--- /dev/null
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# 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.
+"""
+Hard-coded information about tag catagories.
+
+These tables have been copied verbatim from the old PHP code. For future
+version a more flexible formatting is required.
+"""
+
+ICONS = {
+ ('boundary', 'administrative'): 'poi_boundary_administrative',
+ ('place', 'city'): 'poi_place_city',
+ ('place', 'town'): 'poi_place_town',
+ ('place', 'village'): 'poi_place_village',
+ ('place', 'hamlet'): 'poi_place_village',
+ ('place', 'suburb'): 'poi_place_village',
+ ('place', 'locality'): 'poi_place_village',
+ ('place', 'airport'): 'transport_airport2',
+ ('aeroway', 'aerodrome'): 'transport_airport2',
+ ('railway', 'station'): 'transport_train_station2',
+ ('amenity', 'place_of_worship'): 'place_of_worship_unknown3',
+ ('amenity', 'pub'): 'food_pub',
+ ('amenity', 'bar'): 'food_bar',
+ ('amenity', 'university'): 'education_university',
+ ('tourism', 'museum'): 'tourist_museum',
+ ('amenity', 'arts_centre'): 'tourist_art_gallery2',
+ ('tourism', 'zoo'): 'tourist_zoo',
+ ('tourism', 'theme_park'): 'poi_point_of_interest',
+ ('tourism', 'attraction'): 'poi_point_of_interest',
+ ('leisure', 'golf_course'): 'sport_golf',
+ ('historic', 'castle'): 'tourist_castle',
+ ('amenity', 'hospital'): 'health_hospital',
+ ('amenity', 'school'): 'education_school',
+ ('amenity', 'theatre'): 'tourist_theatre',
+ ('amenity', 'library'): 'amenity_library',
+ ('amenity', 'fire_station'): 'amenity_firestation3',
+ ('amenity', 'police'): 'amenity_police2',
+ ('amenity', 'bank'): 'money_bank2',
+ ('amenity', 'post_office'): 'amenity_post_office',
+ ('tourism', 'hotel'): 'accommodation_hotel2',
+ ('amenity', 'cinema'): 'tourist_cinema',
+ ('tourism', 'artwork'): 'tourist_art_gallery2',
+ ('historic', 'archaeological_site'): 'tourist_archaeological2',
+ ('amenity', 'doctors'): 'health_doctors',
+ ('leisure', 'sports_centre'): 'sport_leisure_centre',
+ ('leisure', 'swimming_pool'): 'sport_swimming_outdoor',
+ ('shop', 'supermarket'): 'shopping_supermarket',
+ ('shop', 'convenience'): 'shopping_convenience',
+ ('amenity', 'restaurant'): 'food_restaurant',
+ ('amenity', 'fast_food'): 'food_fastfood',
+ ('amenity', 'cafe'): 'food_cafe',
+ ('tourism', 'guest_house'): 'accommodation_bed_and_breakfast',
+ ('amenity', 'pharmacy'): 'health_pharmacy_dispensing',
+ ('amenity', 'fuel'): 'transport_fuel',
+ ('natural', 'peak'): 'poi_peak',
+ ('natural', 'wood'): 'landuse_coniferous_and_deciduous',
+ ('shop', 'bicycle'): 'shopping_bicycle',
+ ('shop', 'clothes'): 'shopping_clothes',
+ ('shop', 'hairdresser'): 'shopping_hairdresser',
+ ('shop', 'doityourself'): 'shopping_diy',
+ ('shop', 'estate_agent'): 'shopping_estateagent2',
+ ('shop', 'car'): 'shopping_car',
+ ('shop', 'garden_centre'): 'shopping_garden_centre',
+ ('shop', 'car_repair'): 'shopping_car_repair',
+ ('shop', 'bakery'): 'shopping_bakery',
+ ('shop', 'butcher'): 'shopping_butcher',
+ ('shop', 'apparel'): 'shopping_clothes',
+ ('shop', 'laundry'): 'shopping_laundrette',
+ ('shop', 'beverages'): 'shopping_alcohol',
+ ('shop', 'alcohol'): 'shopping_alcohol',
+ ('shop', 'optician'): 'health_opticians',
+ ('shop', 'chemist'): 'health_pharmacy',
+ ('shop', 'gallery'): 'tourist_art_gallery2',
+ ('shop', 'jewelry'): 'shopping_jewelry',
+ ('tourism', 'information'): 'amenity_information',
+ ('historic', 'ruins'): 'tourist_ruin',
+ ('amenity', 'college'): 'education_school',
+ ('historic', 'monument'): 'tourist_monument',
+ ('historic', 'memorial'): 'tourist_monument',
+ ('historic', 'mine'): 'poi_mine',
+ ('tourism', 'caravan_site'): 'accommodation_caravan_park',
+ ('amenity', 'bus_station'): 'transport_bus_station',
+ ('amenity', 'atm'): 'money_atm2',
+ ('tourism', 'viewpoint'): 'tourist_view_point',
+ ('tourism', 'guesthouse'): 'accommodation_bed_and_breakfast',
+ ('railway', 'tram'): 'transport_tram_stop',
+ ('amenity', 'courthouse'): 'amenity_court',
+ ('amenity', 'recycling'): 'amenity_recycling',
+ ('amenity', 'dentist'): 'health_dentist',
+ ('natural', 'beach'): 'tourist_beach',
+ ('railway', 'tram_stop'): 'transport_tram_stop',
+ ('amenity', 'prison'): 'amenity_prison',
+ ('highway', 'bus_stop'): 'transport_bus_stop2'
+}
"""
Output formatters for API version v1.
"""
+from typing import Mapping, Any
+import collections
+
+import nominatim.api as napi
from nominatim.api.result_formatting import FormatDispatcher
-from nominatim.api import StatusResult
+from nominatim.api.v1.classtypes import ICONS
from nominatim.utils.json_writer import JsonWriter
dispatch = FormatDispatcher()
-@dispatch.format_func(StatusResult, 'text')
-def _format_status_text(result: StatusResult) -> str:
+@dispatch.format_func(napi.StatusResult, 'text')
+def _format_status_text(result: napi.StatusResult, _: Mapping[str, Any]) -> str:
if result.status:
return f"ERROR: {result.message}"
return 'OK'
-@dispatch.format_func(StatusResult, 'json')
-def _format_status_json(result: StatusResult) -> str:
+@dispatch.format_func(napi.StatusResult, 'json')
+def _format_status_json(result: napi.StatusResult, _: Mapping[str, Any]) -> str:
out = JsonWriter()
out.start_object()\
.end_object()
return out()
+
+
+def _add_address_row(writer: JsonWriter, row: napi.AddressLine,
+ locales: napi.Locales) -> None:
+ writer.start_object()\
+ .keyval('localname', locales.display_name(row.names))\
+ .keyval_not_none('place_id', row.place_id)
+
+ if row.osm_object is not None:
+ writer.keyval('osm_id', row.osm_object[1])\
+ .keyval('osm_type', row.osm_object[0])
+
+ if row.extratags:
+ writer.keyval_not_none('place_type', row.extratags.get('place_type'))
+
+ writer.keyval('class', row.category[0])\
+ .keyval('type', row.category[1])\
+ .keyval_not_none('admin_level', row.admin_level)\
+ .keyval('rank_address', row.rank_address)\
+ .keyval('distance', row.distance)\
+ .keyval('isaddress', row.isaddress)\
+ .end_object()
+
+
+def _add_address_rows(writer: JsonWriter, section: str, rows: napi.AddressLines,
+ locales: napi.Locales) -> None:
+ writer.key(section).start_array()
+ for row in rows:
+ _add_address_row(writer, row, locales)
+ writer.next()
+ writer.end_array().next()
+
+
+def _add_parent_rows_grouped(writer: JsonWriter, rows: napi.AddressLines,
+ locales: napi.Locales) -> None:
+ # group by category type
+ data = collections.defaultdict(list)
+ for row in rows:
+ sub = JsonWriter()
+ _add_address_row(sub, row, locales)
+ data[row.category[1]].append(sub())
+
+ writer.key('hierarchy').start_object()
+ for group, grouped in data.items():
+ writer.key(group).start_array()
+ grouped.sort() # sorts alphabetically by local name
+ for line in grouped:
+ writer.raw(line).next()
+ writer.end_array().next()
+
+ writer.end_object().next()
+
+
+@dispatch.format_func(napi.SearchResult, 'details-json')
+def _format_search_json(result: napi.SearchResult, options: Mapping[str, Any]) -> str:
+ locales = options.get('locales', napi.Locales())
+ geom = result.geometry.get('geojson')
+ centroid = result.centroid_as_geojson()
+
+ out = JsonWriter()
+ out.start_object()\
+ .keyval_not_none('place_id', result.place_id)\
+ .keyval_not_none('parent_place_id', result.parent_place_id)
+
+ if result.osm_object is not None:
+ out.keyval('osm_type', result.osm_object[0])\
+ .keyval('osm_id', result.osm_object[1])
+
+ out.keyval('category', result.category[0])\
+ .keyval('type', result.category[1])\
+ .keyval('admin_level', result.admin_level)\
+ .keyval('localname', locales.display_name(result.names))\
+ .keyval_not_none('names', result.names or None)\
+ .keyval_not_none('addresstags', result.address or None)\
+ .keyval_not_none('housenumber', result.housenumber)\
+ .keyval_not_none('calculated_postcode', result.postcode)\
+ .keyval_not_none('country_code', result.country_code)\
+ .keyval_not_none('indexed_date', result.indexed_date, lambda v: v.isoformat())\
+ .keyval_not_none('importance', result.importance)\
+ .keyval('calculated_importance', result.calculated_importance())\
+ .keyval_not_none('extratags', result.extratags or None)\
+ .keyval_not_none('calculated_wikipedia', result.wikipedia)\
+ .keyval('rank_address', result.rank_address)\
+ .keyval('rank_search', result.rank_search)\
+ .keyval('isarea', 'Polygon' in (geom or result.geometry.get('type') or ''))\
+ .key('centroid').raw(centroid).next()\
+ .key('geometry').raw(geom or centroid).next()
+
+ if options.get('icon_base_url', None):
+ icon = ICONS.get(result.category)
+ if icon:
+ out.keyval('icon', f"{options['icon_base_url']}/{icon}.p.20.png")
+
+ if result.address_rows is not None:
+ _add_address_rows(out, 'address', result.address_rows, locales)
+
+ if result.linked_rows is not None:
+ _add_address_rows(out, 'linked_places', result.linked_rows, locales)
+
+ if result.name_keywords is not None or result.address_keywords is not None:
+ out.key('keywords').start_object()
+
+ for sec, klist in (('name', result.name_keywords), ('address', result.address_keywords)):
+ out.key(sec).start_array()
+ for word in (klist or []):
+ out.start_object()\
+ .keyval('id', word.word_id)\
+ .keyval('token', word.word_token)\
+ .end_object().next()
+ out.end_array().next()
+
+ out.end_object().next()
+
+ if result.parented_rows is not None:
+ if options.get('group_hierarchy', False):
+ _add_parent_rows_grouped(out, result.parented_rows, locales)
+ else:
+ _add_address_rows(out, 'hierarchy', result.parented_rows, locales)
+
+ out.end_object()
+
+ return out()
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
else:
status_code = 200
- return params.build_response(formatting.format_result(result, fmt), fmt,
+ 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 typing import Mapping, Dict
import argparse
import logging
+import json
+import sys
from nominatim.tools.exec_utils import run_api_script
from nominatim.errors import UsageError
from nominatim.clicmd.args import NominatimArgs
-from nominatim.api import NominatimAPI, StatusResult
+import nominatim.api as napi
import nominatim.api.v1 as api_output
# Do not repeat documentation of subcommand classes.
('namedetails', 'Include a list of alternative names')
)
-DETAILS_SWITCHES = (
- ('addressdetails', 'Include a breakdown of the address into elements'),
- ('keywords', 'Include a list of name keywords and address keywords'),
- ('linkedplaces', 'Include a details of places that are linked with this one'),
- ('hierarchy', 'Include details of places lower in the address hierarchy'),
- ('group_hierarchy', 'Group the places by type'),
- ('polygon_geojson', 'Include geometry of result')
-)
-
def _add_api_output_arguments(parser: argparse.ArgumentParser) -> None:
group = parser.add_argument_group('Output arguments')
group.add_argument('--format', default='jsonv2',
"of the same object."))
group = parser.add_argument_group('Output arguments')
- for name, desc in DETAILS_SWITCHES:
- group.add_argument('--' + name, action='store_true', help=desc)
+ group.add_argument('--addressdetails', action='store_true',
+ help='Include a breakdown of the address into elements')
+ group.add_argument('--keywords', action='store_true',
+ help='Include a list of name keywords and address keywords')
+ group.add_argument('--linkedplaces', action='store_true',
+ help='Include a details of places that are linked with this one')
+ group.add_argument('--hierarchy', action='store_true',
+ help='Include details of places lower in the address hierarchy')
+ group.add_argument('--group_hierarchy', action='store_true',
+ help='Group the places by type')
+ group.add_argument('--polygon_geojson', action='store_true',
+ help='Include geometry of result')
group.add_argument('--lang', '--accept-language', metavar='LANGS',
help='Preferred language order for presenting search results')
def run(self, args: NominatimArgs) -> int:
+ place: napi.PlaceRef
if args.node:
- params = dict(osmtype='N', osmid=args.node)
+ place = napi.OsmID('N', args.node, args.object_class)
elif args.way:
- params = dict(osmtype='W', osmid=args.way)
+ place = napi.OsmID('W', args.way, args.object_class)
elif args.relation:
- params = dict(osmtype='R', osmid=args.relation)
+ place = napi.OsmID('R', args.relation, args.object_class)
else:
- params = dict(place_id=args.place_id)
- if args.object_class:
- params['class'] = args.object_class
- for name, _ in DETAILS_SWITCHES:
- params[name] = '1' if getattr(args, name) else '0'
+ assert args.place_id is not None
+ place = napi.PlaceID(args.place_id)
+
+ api = napi.NominatimAPI(args.project_dir)
+
+ details = napi.LookupDetails(address_details=args.addressdetails,
+ linked_places=args.linkedplaces,
+ parented_places=args.hierarchy,
+ keywords=args.keywords)
+ if args.polygon_geojson:
+ details.geometry_output = napi.GeometryFormat.GEOJSON
+
if args.lang:
- params['accept-language'] = args.lang
+ locales = napi.Locales.from_accept_languages(args.lang)
+ elif api.config.DEFAULT_LANGUAGE:
+ locales = napi.Locales.from_accept_languages(api.config.DEFAULT_LANGUAGE)
+ else:
+ locales = napi.Locales()
+
+ result = api.lookup(place, details)
+
+ if result:
+ output = api_output.format_result(
+ result,
+ 'details-json',
+ {'locales': locales,
+ 'group_hierarchy': args.group_hierarchy})
+ # reformat the result, so it is pretty-printed
+ json.dump(json.loads(output), sys.stdout, indent=4)
+ sys.stdout.write('\n')
+
+ return 0
- return _run_api('details', args, params)
+ LOG.error("Object not found in database.")
+ return 42
class APIStatus:
"""
def add_args(self, parser: argparse.ArgumentParser) -> None:
- formats = api_output.list_formats(StatusResult)
+ formats = api_output.list_formats(napi.StatusResult)
group = parser.add_argument_group('API parameters')
group.add_argument('--format', default=formats[0], choices=formats,
help='Format of result')
def run(self, args: NominatimArgs) -> int:
- status = NominatimAPI(args.project_dir).status()
- print(api_output.format_result(status, args.format))
+ status = napi.NominatimAPI(args.project_dir).status()
+ print(api_output.format_result(status, args.format, {}))
return 0
# Arguments to 'details'
object_class: Optional[str]
+ linkedplaces: bool
+ hierarchy: bool
+ keywords: bool
+ polygon_geojson: bool
+ group_hierarchy: bool
def osm2pgsql_options(self, default_cache: int,
from sqlalchemy.dialects.postgresql import HSTORE, ARRAY, JSONB
from sqlalchemy.dialects.sqlite import JSON as sqlite_json
+class PostgresTypes:
+ """ Type definitions for complex types as used in Postgres variants.
+ """
+ Composite = HSTORE
+ Json = JSONB
+ IntArray = ARRAY(sa.Integer()) #pylint: disable=invalid-name
+
+
+class SqliteTypes:
+ """ Type definitions for complex types as used in Postgres variants.
+ """
+ Composite = sqlite_json
+ Json = sqlite_json
+ IntArray = sqlite_json
+
+
#pylint: disable=too-many-instance-attributes
class SearchTables:
""" Data class that holds the tables of the Nominatim database.
def __init__(self, meta: sa.MetaData, engine_name: str) -> None:
if engine_name == 'postgresql':
- Composite: Any = HSTORE
- Json: Any = JSONB
- IntArray: Any = ARRAY(sa.Integer()) #pylint: disable=invalid-name
+ self.types: Any = PostgresTypes
elif engine_name == 'sqlite':
- Composite = sqlite_json
- Json = sqlite_json
- IntArray = sqlite_json
+ self.types = SqliteTypes
else:
raise ValueError("Only 'postgresql' and 'sqlite' engines are supported.")
sa.Column('class', sa.Text, nullable=False, key='class_'),
sa.Column('type', sa.Text, nullable=False),
sa.Column('admin_level', sa.SmallInteger),
- sa.Column('name', Composite),
- sa.Column('address', Composite),
- sa.Column('extratags', Composite),
+ sa.Column('name', self.types.Composite),
+ sa.Column('address', self.types.Composite),
+ sa.Column('extratags', self.types.Composite),
sa.Column('geometry', Geometry(srid=4326), nullable=False),
sa.Column('wikipedia', sa.Text),
sa.Column('country_code', sa.String(2)),
sa.Column('partition', sa.SmallInteger),
sa.Column('indexed_status', sa.SmallInteger),
sa.Column('linegeo', Geometry(srid=4326)),
- sa.Column('address', Composite),
+ sa.Column('address', self.types.Composite),
sa.Column('postcode', sa.Text),
sa.Column('country_code', sa.String(2)))
sa.Column('word_token', sa.Text, nullable=False),
sa.Column('type', sa.Text, nullable=False),
sa.Column('word', sa.Text),
- sa.Column('info', Json))
+ sa.Column('info', self.types.Json))
self.country_name = sa.Table('country_name', meta,
sa.Column('country_code', sa.String(2)),
- sa.Column('name', Composite),
- sa.Column('derived_name', Composite),
+ sa.Column('name', self.types.Composite),
+ sa.Column('derived_name', self.types.Composite),
sa.Column('country_default_language_code', sa.Text),
sa.Column('partition', sa.Integer))
sa.Column('importance', sa.Float),
sa.Column('search_rank', sa.SmallInteger),
sa.Column('address_rank', sa.SmallInteger),
- sa.Column('name_vector', IntArray, index=True),
- sa.Column('nameaddress_vector', IntArray, index=True),
+ sa.Column('name_vector', self.types.IntArray, index=True),
+ sa.Column('nameaddress_vector', self.types.IntArray, index=True),
sa.Column('country_code', sa.String(2)),
sa.Column('centroid', Geometry(srid=4326)))
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:
there.
"""
def decorator(func: Callable[..., None]) -> Callable[..., None]:
- version = (NominatimVersion(major, minor, patch, dbpatch))
+ version = NominatimVersion(major, minor, patch, dbpatch)
_MIGRATION_FUNCTIONS.append((version, func))
return func
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2022 by the Nominatim developer community.
+# Copyright (C) 2023 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Type definitions for typing annotations.
Protocol = object
Final = 'Final'
TypedDict = dict
+
+
+# SQLAlchemy introduced generic types in version 2.0 making typing
+# inclompatiple with older versions. Add wrappers here so we don't have
+# to litter the code with bare-string types.
+
+if TYPE_CHECKING:
+ import sqlalchemy as sa
+ from typing_extensions import (TypeAlias as TypeAlias)
+else:
+ TypeAlias = str
+
+SaSelect: TypeAlias = 'sa.Select[Any]'
+SaRow: TypeAlias = 'sa.Row[Any]'
+SaColumn: TypeAlias = 'sa.Column[Any]'
+SaLabel: TypeAlias = 'sa.Label[Any]'
Then the result is valid json
And result has attributes geometry
And result has not attributes keywords,address,linked_places,parentof
+ And results contain
+ | geometry+type |
+ | Point |
Scenario: JSON Details with pretty printing
When sending json details query for W297699560
| keywords |
| 1 |
Then the result is valid json
+ And result has attributes keywords
Scenario Outline: JSON details with full geometry
When sending json details query for <osmid>
| 1 |
Then the result is valid json
And result has attributes geometry
+ And results contain
+ | geometry+type |
+ | <geometry> |
Examples:
- | osmid |
- | W297699560 |
- | W243055645 |
- | W243055716 |
- | W43327921 |
+ | osmid | geometry |
+ | W297699560 | LineString |
+ | W243055645 | Polygon |
+ | W243055716 | Polygon |
+ | W43327921 | LineString |
| place_id |
| 107077 |
+
Scenario Outline: Details via OSM id
When sending details query for <type><id>
Then the result is valid json
| W | 43327921 |
| R | 123924 |
- Scenario: Details for interpolation way just return the dependent street
- When sending details query for W1
- Then the result is valid json
- And results contain
- | category |
- | highway |
Scenario Outline: Details for different class types for the same OSM id
When sending details query for N300209696:<class>
| natural |
| mountain_pass |
+
Scenario Outline: Details via unknown OSM id
When sending details query for <object>
Then a HTTP 404 is returned
| N300209696:highway |
+ @v1-api-php-only
+ Scenario: Details for interpolation way just return the dependent street
+ When sending details query for W1
+ Then the result is valid json
+ And results contain
+ | category |
+ | highway |
+
+
+ @v1-api-python-only
+ Scenario: Details for interpolation way return the interpolation
+ When sending details query for W1
+ Then the result is valid json
+ And results contain
+ | category | type | osm_type | osm_id | admin_level |
+ | place | houses | W | 1 | 15 |
+
+
+ @v1-api-php-only
+ Scenario: Details for Tiger way just return the dependent street
+ When sending details query for 112871
+ Then the result is valid json
+ And results contain
+ | category |
+ | highway |
+
+
+ @v1-api-python-only
+ Scenario: Details for interpolation way return the interpolation
+ When sending details query for 112871
+ Then the result is valid json
+ And results contain
+ | category | type | admin_level |
+ | place | houses | 15 |
+ And result has not attributes osm_type,osm_id
+
+
+ @v1-api-php-only
+ Scenario: Details for postcodes just return the dependent place
+ When sending details query for 112820
+ Then the result is valid json
+ And results contain
+ | category |
+ | boundary |
+
+
+ @v1-api-python-only
+ Scenario: Details for interpolation way return the interpolation
+ When sending details query for 112820
+ Then the result is valid json
+ And results contain
+ | category | type | admin_level |
+ | place | postcode | 15 |
+ And result has not attributes osm_type,osm_id
if tag == 'fail-legacy':
if context.config.userdata['TOKENIZER'] == 'legacy':
context.scenario.skip("Not implemented in legacy tokenizer")
+ if tag == 'v1-api-php-only':
+ if context.config.userdata['API_ENGINE'] != 'php':
+ context.scenario.skip("Only valid with PHP version of v1 API.")
+ if tag == 'v1-api-python-only':
+ if context.config.userdata['API_ENGINE'] == 'php':
+ context.scenario.skip("Only valid with Python version of v1 API.")
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2022 by the Nominatim developer community.
+# Copyright (C) 2023 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Classes wrapping HTTP responses from the Nominatim API.
assert str(self.result[idx][field]) == str(value), \
BadRowValueAssert(self, idx, field, value)
+
+ def assert_subfield(self, idx, path, value):
+ assert path
+
+ field = self.result[idx]
+ for p in path:
+ assert isinstance(field, OrderedDict)
+ assert p in field
+ field = field[p]
+
+ if isinstance(value, float):
+ assert Almost(value) == float(field)
+ elif value.startswith("^"):
+ assert re.fullmatch(value, field)
+ elif isinstance(field, OrderedDict):
+ assert field, eval('{' + value + '}')
+ else:
+ assert str(field) == str(value)
+
+
def assert_address_field(self, idx, field, value):
""" Check that result rows`idx` has a field `field` with value `value`
in its address. If idx is None, then all results are checked.
raise RuntimeError("Context needed when using grid coordinates")
self.assert_field(i, 'lat', float(lat))
self.assert_field(i, 'lon', float(lon))
+ elif '+' in name:
+ self.assert_subfield(i, name.split('+'), value)
else:
self.assert_field(i, name, value)
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', {})))
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2022 by the Nominatim developer community.
+# Copyright (C) 2023 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Helper fixtures for API call tests.
from pathlib import Path
import pytest
import time
+import datetime as dt
+
+import nominatim.api as napi
+from nominatim.db.sql_preprocessor import SQLPreprocessor
+
+class APITester:
+
+ def __init__(self):
+ self.api = napi.NominatimAPI(Path('/invalid'))
+ self.async_to_sync(self.api._async_api.setup_database())
+
+
+ def async_to_sync(self, func):
+ """ Run an asynchronous function until completion using the
+ internal loop of the API.
+ """
+ return self.api._loop.run_until_complete(func)
+
+
+ def add_data(self, table, data):
+ """ Insert data into the given table.
+ """
+ sql = getattr(self.api._async_api._tables, table).insert()
+ self.async_to_sync(self.exec_async(sql, data))
+
+
+ def add_placex(self, **kw):
+ name = kw.get('name')
+ if isinstance(name, str):
+ name = {'name': name}
+
+ self.add_data('placex',
+ {'place_id': kw.get('place_id', 1000),
+ 'osm_type': kw.get('osm_type', 'W'),
+ 'osm_id': kw.get('osm_id', 4),
+ 'class_': kw.get('class_', 'highway'),
+ 'type': kw.get('type', 'residential'),
+ 'name': name,
+ 'address': kw.get('address'),
+ 'extratags': kw.get('extratags'),
+ 'parent_place_id': kw.get('parent_place_id'),
+ 'linked_place_id': kw.get('linked_place_id'),
+ 'admin_level': kw.get('admin_level', 15),
+ 'country_code': kw.get('country_code'),
+ 'housenumber': kw.get('housenumber'),
+ 'postcode': kw.get('postcode'),
+ 'wikipedia': kw.get('wikipedia'),
+ 'rank_search': kw.get('rank_search', 30),
+ 'rank_address': kw.get('rank_address', 30),
+ 'importance': kw.get('importance'),
+ 'centroid': 'SRID=4326;POINT(%f %f)' % kw.get('centroid', (23.0, 34.0)),
+ 'indexed_date': kw.get('indexed_date',
+ dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
+ 'geometry': 'SRID=4326;' + kw.get('geometry', 'POINT(23 34)')})
+
+
+ def add_address_placex(self, object_id, **kw):
+ self.add_placex(**kw)
+ self.add_data('addressline',
+ {'place_id': object_id,
+ 'address_place_id': kw.get('place_id', 1000),
+ 'distance': kw.get('distance', 0.0),
+ 'cached_rank_address': kw.get('rank_address', 30),
+ 'fromarea': kw.get('fromarea', False),
+ 'isaddress': kw.get('isaddress', True)})
+
+
+ def add_osmline(self, **kw):
+ self.add_data('osmline',
+ {'place_id': kw.get('place_id', 10000),
+ 'osm_id': kw.get('osm_id', 4004),
+ 'parent_place_id': kw.get('parent_place_id'),
+ 'indexed_date': kw.get('indexed_date',
+ dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
+ 'startnumber': kw.get('startnumber', 2),
+ 'endnumber': kw.get('endnumber', 6),
+ 'step': kw.get('step', 2),
+ 'address': kw.get('address'),
+ 'postcode': kw.get('postcode'),
+ 'country_code': kw.get('country_code'),
+ 'linegeo': 'SRID=4326;' + kw.get('geometry', 'LINESTRING(1.1 -0.2, 1.09 -0.22)')})
+
+
+ def add_tiger(self, **kw):
+ self.add_data('tiger',
+ {'place_id': kw.get('place_id', 30000),
+ 'parent_place_id': kw.get('parent_place_id'),
+ 'startnumber': kw.get('startnumber', 2),
+ 'endnumber': kw.get('endnumber', 6),
+ 'step': kw.get('step', 2),
+ 'postcode': kw.get('postcode'),
+ 'linegeo': 'SRID=4326;' + kw.get('geometry', 'LINESTRING(1.1 -0.2, 1.09 -0.22)')})
+
+
+ def add_postcode(self, **kw):
+ self.add_data('postcode',
+ {'place_id': kw.get('place_id', 1000),
+ 'parent_place_id': kw.get('parent_place_id'),
+ 'country_code': kw.get('country_code'),
+ 'postcode': kw.get('postcode'),
+ 'rank_search': kw.get('rank_search', 20),
+ 'rank_address': kw.get('rank_address', 22),
+ 'indexed_date': kw.get('indexed_date',
+ dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
+ 'geometry': 'SRID=4326;' + kw.get('geometry', 'POINT(23 34)')})
+
+
+ async def exec_async(self, sql, *args, **kwargs):
+ async with self.api._async_api.begin() as conn:
+ return await conn.execute(sql, *args, **kwargs)
+
+
+ async def create_tables(self):
+ async with self.api._async_api._engine.begin() as conn:
+ await conn.run_sync(self.api._async_api._tables.meta.create_all)
-from nominatim.api import NominatimAPI
@pytest.fixture
-def apiobj(temp_db):
+def apiobj(temp_db_with_extensions, temp_db_conn, monkeypatch):
""" Create an asynchronous SQLAlchemy engine for the test DB.
"""
- api = NominatimAPI(Path('/invalid'), {})
- yield api
- api.close()
+ monkeypatch.setenv('NOMINATIM_USE_US_TIGER_DATA', 'yes')
+ testapi = APITester()
+ testapi.async_to_sync(testapi.create_tables())
+
+ SQLPreprocessor(temp_db_conn, testapi.api.config)\
+ .run_sql_file(temp_db_conn, 'functions/address_lookup.sql')
+
+ yield testapi
+
+ testapi.api.close()
--- /dev/null
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# 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.
+"""
+Tests for lookup API call.
+"""
+import datetime as dt
+
+import pytest
+
+import nominatim.api as napi
+
+@pytest.mark.parametrize('idobj', (napi.PlaceID(332), napi.OsmID('W', 4),
+ napi.OsmID('W', 4, 'highway')))
+def test_lookup_in_placex(apiobj, idobj):
+ import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
+ apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
+ class_='highway', type='residential',
+ name={'name': 'Road'}, address={'city': 'Barrow'},
+ extratags={'surface': 'paved'},
+ parent_place_id=34, linked_place_id=55,
+ admin_level=15, country_code='gb',
+ housenumber='4',
+ postcode='34425', wikipedia='en:Faa',
+ rank_search=27, rank_address=26,
+ importance=0.01,
+ centroid=(23, 34),
+ indexed_date=import_date,
+ geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
+
+ result = apiobj.api.lookup(idobj, napi.LookupDetails())
+
+ assert result is not None
+
+ assert result.source_table.name == 'PLACEX'
+ assert result.category == ('highway', 'residential')
+ assert result.centroid == (pytest.approx(23.0), pytest.approx(34.0))
+
+ assert result.place_id == 332
+ assert result.parent_place_id == 34
+ assert result.linked_place_id == 55
+ assert result.osm_object == ('W', 4)
+ assert result.admin_level == 15
+
+ assert result.names == {'name': 'Road'}
+ assert result.address == {'city': 'Barrow'}
+ assert result.extratags == {'surface': 'paved'}
+
+ assert result.housenumber == '4'
+ assert result.postcode == '34425'
+ assert result.wikipedia == 'en:Faa'
+
+ assert result.rank_search == 27
+ assert result.rank_address == 26
+ assert result.importance == pytest.approx(0.01)
+
+ assert result.country_code == 'gb'
+ assert result.indexed_date == import_date.replace(tzinfo=dt.timezone.utc)
+
+ assert result.address_rows is None
+ assert result.linked_rows is None
+ assert result.parented_rows is None
+ assert result.name_keywords is None
+ assert result.address_keywords is None
+
+ assert result.geometry == {'type': 'ST_LineString'}
+
+
+def test_lookup_in_placex_minimal_info(apiobj):
+ import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
+ apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
+ class_='highway', type='residential',
+ admin_level=15,
+ rank_search=27, rank_address=26,
+ centroid=(23, 34),
+ indexed_date=import_date,
+ geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
+
+ result = apiobj.api.lookup(napi.PlaceID(332), napi.LookupDetails())
+
+ assert result is not None
+
+ assert result.source_table.name == 'PLACEX'
+ assert result.category == ('highway', 'residential')
+ assert result.centroid == (pytest.approx(23.0), pytest.approx(34.0))
+
+ assert result.place_id == 332
+ assert result.parent_place_id is None
+ assert result.linked_place_id is None
+ assert result.osm_object == ('W', 4)
+ assert result.admin_level == 15
+
+ assert result.names is None
+ assert result.address is None
+ assert result.extratags is None
+
+ assert result.housenumber is None
+ assert result.postcode is None
+ assert result.wikipedia is None
+
+ assert result.rank_search == 27
+ assert result.rank_address == 26
+ assert result.importance is None
+
+ assert result.country_code is None
+ assert result.indexed_date == import_date.replace(tzinfo=dt.timezone.utc)
+
+ assert result.address_rows is None
+ assert result.linked_rows is None
+ assert result.parented_rows is None
+ assert result.name_keywords is None
+ assert result.address_keywords is None
+
+ assert result.geometry == {'type': 'ST_LineString'}
+
+
+def test_lookup_in_placex_with_geometry(apiobj):
+ apiobj.add_placex(place_id=332,
+ geometry='LINESTRING(23 34, 23.1 34)')
+
+ result = apiobj.api.lookup(napi.PlaceID(332),
+ napi.LookupDetails(geometry_output=napi.GeometryFormat.GEOJSON))
+
+ assert result.geometry == {'geojson': '{"type":"LineString","coordinates":[[23,34],[23.1,34]]}'}
+
+
+def test_lookup_placex_with_address_details(apiobj):
+ apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
+ class_='highway', type='residential', name='Street',
+ country_code='pl',
+ rank_search=27, rank_address=26)
+ apiobj.add_address_placex(332, fromarea=False, isaddress=False,
+ distance=0.0034,
+ place_id=1000, osm_type='N', osm_id=3333,
+ class_='place', type='suburb', name='Smallplace',
+ country_code='pl', admin_level=13,
+ rank_search=24, rank_address=23)
+ apiobj.add_address_placex(332, fromarea=True, isaddress=True,
+ place_id=1001, osm_type='N', osm_id=3334,
+ class_='place', type='city', name='Bigplace',
+ country_code='pl',
+ rank_search=17, rank_address=16)
+
+ result = apiobj.api.lookup(napi.PlaceID(332),
+ napi.LookupDetails(address_details=True))
+
+ assert result.address_rows == [
+ napi.AddressLine(place_id=332, osm_object=('W', 4),
+ category=('highway', 'residential'),
+ names={'name': 'Street'}, extratags={},
+ admin_level=15, fromarea=True, isaddress=True,
+ rank_address=26, distance=0.0),
+ napi.AddressLine(place_id=1000, osm_object=('N', 3333),
+ category=('place', 'suburb'),
+ names={'name': 'Smallplace'}, extratags={},
+ admin_level=13, fromarea=False, isaddress=True,
+ rank_address=23, distance=0.0034),
+ napi.AddressLine(place_id=1001, osm_object=('N', 3334),
+ category=('place', 'city'),
+ names={'name': 'Bigplace'}, extratags={},
+ admin_level=15, fromarea=True, isaddress=True,
+ rank_address=16, distance=0.0),
+ napi.AddressLine(place_id=None, osm_object=None,
+ category=('place', 'country_code'),
+ names={'ref': 'pl'}, extratags={},
+ admin_level=None, fromarea=True, isaddress=False,
+ rank_address=4, distance=0.0)
+ ]
+
+
+def test_lookup_place_with_linked_places_none_existing(apiobj):
+ apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
+ class_='highway', type='residential', name='Street',
+ country_code='pl', linked_place_id=45,
+ rank_search=27, rank_address=26)
+
+ result = apiobj.api.lookup(napi.PlaceID(332),
+ napi.LookupDetails(linked_places=True))
+
+ assert result.linked_rows == []
+
+
+def test_lookup_place_with_linked_places_existing(apiobj):
+ apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
+ class_='highway', type='residential', name='Street',
+ country_code='pl', linked_place_id=45,
+ rank_search=27, rank_address=26)
+ apiobj.add_placex(place_id=1001, osm_type='W', osm_id=5,
+ class_='highway', type='residential', name='Street',
+ country_code='pl', linked_place_id=332,
+ rank_search=27, rank_address=26)
+ apiobj.add_placex(place_id=1002, osm_type='W', osm_id=6,
+ class_='highway', type='residential', name='Street',
+ country_code='pl', linked_place_id=332,
+ rank_search=27, rank_address=26)
+
+ result = apiobj.api.lookup(napi.PlaceID(332),
+ napi.LookupDetails(linked_places=True))
+
+ assert result.linked_rows == [
+ napi.AddressLine(place_id=1001, osm_object=('W', 5),
+ category=('highway', 'residential'),
+ names={'name': 'Street'}, extratags={},
+ admin_level=15, fromarea=False, isaddress=True,
+ rank_address=26, distance=0.0),
+ napi.AddressLine(place_id=1002, osm_object=('W', 6),
+ category=('highway', 'residential'),
+ names={'name': 'Street'}, extratags={},
+ admin_level=15, fromarea=False, isaddress=True,
+ rank_address=26, distance=0.0),
+ ]
+
+
+def test_lookup_place_with_parented_places_not_existing(apiobj):
+ apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
+ class_='highway', type='residential', name='Street',
+ country_code='pl', parent_place_id=45,
+ rank_search=27, rank_address=26)
+
+ result = apiobj.api.lookup(napi.PlaceID(332),
+ napi.LookupDetails(parented_places=True))
+
+ assert result.parented_rows == []
+
+
+def test_lookup_place_with_parented_places_existing(apiobj):
+ apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
+ class_='highway', type='residential', name='Street',
+ country_code='pl', parent_place_id=45,
+ rank_search=27, rank_address=26)
+ apiobj.add_placex(place_id=1001, osm_type='N', osm_id=5,
+ class_='place', type='house', housenumber='23',
+ country_code='pl', parent_place_id=332,
+ rank_search=30, rank_address=30)
+ apiobj.add_placex(place_id=1002, osm_type='W', osm_id=6,
+ class_='highway', type='residential', name='Street',
+ country_code='pl', parent_place_id=332,
+ rank_search=27, rank_address=26)
+
+ result = apiobj.api.lookup(napi.PlaceID(332),
+ napi.LookupDetails(parented_places=True))
+
+ assert result.parented_rows == [
+ napi.AddressLine(place_id=1001, osm_object=('N', 5),
+ category=('place', 'house'),
+ names={'housenumber': '23'}, extratags={},
+ admin_level=15, fromarea=False, isaddress=True,
+ rank_address=30, distance=0.0),
+ ]
+
+
+@pytest.mark.parametrize('idobj', (napi.PlaceID(4924), napi.OsmID('W', 9928)))
+def test_lookup_in_osmline(apiobj, idobj):
+ import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
+ apiobj.add_osmline(place_id=4924, osm_id=9928,
+ parent_place_id=12,
+ startnumber=1, endnumber=4, step=1,
+ country_code='gb', postcode='34425',
+ address={'city': 'Big'},
+ indexed_date=import_date,
+ geometry='LINESTRING(23 34, 23 35)')
+
+ result = apiobj.api.lookup(idobj, napi.LookupDetails())
+
+ assert result is not None
+
+ assert result.source_table.name == 'OSMLINE'
+ assert result.category == ('place', 'houses')
+ assert result.centroid == (pytest.approx(23.0), pytest.approx(34.5))
+
+ assert result.place_id == 4924
+ assert result.parent_place_id == 12
+ assert result.linked_place_id is None
+ assert result.osm_object == ('W', 9928)
+ assert result.admin_level == 15
+
+ assert result.names is None
+ assert result.address == {'city': 'Big'}
+ assert result.extratags == {'startnumber': '1', 'endnumber': '4', 'step': '1'}
+
+ assert result.housenumber is None
+ assert result.postcode == '34425'
+ assert result.wikipedia is None
+
+ assert result.rank_search == 30
+ assert result.rank_address == 30
+ assert result.importance is None
+
+ assert result.country_code == 'gb'
+ assert result.indexed_date == import_date.replace(tzinfo=dt.timezone.utc)
+
+ assert result.address_rows is None
+ assert result.linked_rows is None
+ assert result.parented_rows is None
+ assert result.name_keywords is None
+ assert result.address_keywords is None
+
+ assert result.geometry == {'type': 'ST_LineString'}
+
+
+def test_lookup_in_osmline_split_interpolation(apiobj):
+ apiobj.add_osmline(place_id=1000, osm_id=9,
+ startnumber=2, endnumber=4, step=1)
+ apiobj.add_osmline(place_id=1001, osm_id=9,
+ startnumber=6, endnumber=9, step=1)
+ apiobj.add_osmline(place_id=1002, osm_id=9,
+ startnumber=11, endnumber=20, step=1)
+
+ for i in range(1, 6):
+ result = apiobj.api.lookup(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
+ assert result.place_id == 1000
+ for i in range(7, 11):
+ result = apiobj.api.lookup(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
+ assert result.place_id == 1001
+ for i in range(12, 22):
+ result = apiobj.api.lookup(napi.OsmID('W', 9, str(i)), napi.LookupDetails())
+ assert result.place_id == 1002
+
+
+def test_lookup_osmline_with_address_details(apiobj):
+ apiobj.add_osmline(place_id=9000, osm_id=9,
+ startnumber=2, endnumber=4, step=1,
+ parent_place_id=332)
+ apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
+ class_='highway', type='residential', name='Street',
+ country_code='pl',
+ rank_search=27, rank_address=26)
+ apiobj.add_address_placex(332, fromarea=False, isaddress=False,
+ distance=0.0034,
+ place_id=1000, osm_type='N', osm_id=3333,
+ class_='place', type='suburb', name='Smallplace',
+ country_code='pl', admin_level=13,
+ rank_search=24, rank_address=23)
+ apiobj.add_address_placex(332, fromarea=True, isaddress=True,
+ place_id=1001, osm_type='N', osm_id=3334,
+ class_='place', type='city', name='Bigplace',
+ country_code='pl',
+ rank_search=17, rank_address=16)
+
+ result = apiobj.api.lookup(napi.PlaceID(9000),
+ napi.LookupDetails(address_details=True))
+
+ assert result.address_rows == [
+ napi.AddressLine(place_id=None, osm_object=None,
+ category=('place', 'house_number'),
+ names={'ref': '2'}, extratags={},
+ admin_level=None, fromarea=True, isaddress=True,
+ rank_address=28, distance=0.0),
+ napi.AddressLine(place_id=332, osm_object=('W', 4),
+ category=('highway', 'residential'),
+ names={'name': 'Street'}, extratags={},
+ admin_level=15, fromarea=True, isaddress=True,
+ rank_address=26, distance=0.0),
+ napi.AddressLine(place_id=1000, osm_object=('N', 3333),
+ category=('place', 'suburb'),
+ names={'name': 'Smallplace'}, extratags={},
+ admin_level=13, fromarea=False, isaddress=True,
+ rank_address=23, distance=0.0034),
+ napi.AddressLine(place_id=1001, osm_object=('N', 3334),
+ category=('place', 'city'),
+ names={'name': 'Bigplace'}, extratags={},
+ admin_level=15, fromarea=True, isaddress=True,
+ rank_address=16, distance=0.0),
+ napi.AddressLine(place_id=None, osm_object=None,
+ category=('place', 'country_code'),
+ names={'ref': 'pl'}, extratags={},
+ admin_level=None, fromarea=True, isaddress=False,
+ rank_address=4, distance=0.0)
+ ]
+
+
+def test_lookup_in_tiger(apiobj):
+ apiobj.add_tiger(place_id=4924,
+ parent_place_id=12,
+ startnumber=1, endnumber=4, step=1,
+ postcode='34425',
+ geometry='LINESTRING(23 34, 23 35)')
+
+ result = apiobj.api.lookup(napi.PlaceID(4924), napi.LookupDetails())
+
+ assert result is not None
+
+ assert result.source_table.name == 'TIGER'
+ assert result.category == ('place', 'houses')
+ assert result.centroid == (pytest.approx(23.0), pytest.approx(34.5))
+
+ assert result.place_id == 4924
+ assert result.parent_place_id == 12
+ assert result.linked_place_id is None
+ assert result.osm_object is None
+ assert result.admin_level == 15
+
+ assert result.names is None
+ assert result.address is None
+ assert result.extratags == {'startnumber': '1', 'endnumber': '4', 'step': '1'}
+
+ assert result.housenumber is None
+ assert result.postcode == '34425'
+ assert result.wikipedia is None
+
+ assert result.rank_search == 30
+ assert result.rank_address == 30
+ assert result.importance is None
+
+ assert result.country_code == 'us'
+ assert result.indexed_date is None
+
+ assert result.address_rows is None
+ assert result.linked_rows is None
+ assert result.parented_rows is None
+ assert result.name_keywords is None
+ assert result.address_keywords is None
+
+ assert result.geometry == {'type': 'ST_LineString'}
+
+
+def test_lookup_tiger_with_address_details(apiobj):
+ apiobj.add_tiger(place_id=9000,
+ startnumber=2, endnumber=4, step=1,
+ parent_place_id=332)
+ apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
+ class_='highway', type='residential', name='Street',
+ country_code='us',
+ rank_search=27, rank_address=26)
+ apiobj.add_address_placex(332, fromarea=False, isaddress=False,
+ distance=0.0034,
+ place_id=1000, osm_type='N', osm_id=3333,
+ class_='place', type='suburb', name='Smallplace',
+ country_code='us', admin_level=13,
+ rank_search=24, rank_address=23)
+ apiobj.add_address_placex(332, fromarea=True, isaddress=True,
+ place_id=1001, osm_type='N', osm_id=3334,
+ class_='place', type='city', name='Bigplace',
+ country_code='us',
+ rank_search=17, rank_address=16)
+
+ result = apiobj.api.lookup(napi.PlaceID(9000),
+ napi.LookupDetails(address_details=True))
+
+ assert result.address_rows == [
+ napi.AddressLine(place_id=None, osm_object=None,
+ category=('place', 'house_number'),
+ names={'ref': '2'}, extratags={},
+ admin_level=None, fromarea=True, isaddress=True,
+ rank_address=28, distance=0.0),
+ napi.AddressLine(place_id=332, osm_object=('W', 4),
+ category=('highway', 'residential'),
+ names={'name': 'Street'}, extratags={},
+ admin_level=15, fromarea=True, isaddress=True,
+ rank_address=26, distance=0.0),
+ napi.AddressLine(place_id=1000, osm_object=('N', 3333),
+ category=('place', 'suburb'),
+ names={'name': 'Smallplace'}, extratags={},
+ admin_level=13, fromarea=False, isaddress=True,
+ rank_address=23, distance=0.0034),
+ napi.AddressLine(place_id=1001, osm_object=('N', 3334),
+ category=('place', 'city'),
+ names={'name': 'Bigplace'}, extratags={},
+ admin_level=15, fromarea=True, isaddress=True,
+ rank_address=16, distance=0.0),
+ napi.AddressLine(place_id=None, osm_object=None,
+ category=('place', 'country_code'),
+ names={'ref': 'us'}, extratags={},
+ admin_level=None, fromarea=True, isaddress=False,
+ rank_address=4, distance=0.0)
+ ]
+
+
+def test_lookup_in_postcode(apiobj):
+ import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
+ apiobj.add_postcode(place_id=554,
+ parent_place_id=152,
+ postcode='34 425',
+ country_code='gb',
+ rank_search=20, rank_address=22,
+ indexed_date=import_date,
+ geometry='POINT(-9.45 5.6)')
+
+ result = apiobj.api.lookup(napi.PlaceID(554), napi.LookupDetails())
+
+ assert result is not None
+
+ assert result.source_table.name == 'POSTCODE'
+ assert result.category == ('place', 'postcode')
+ assert result.centroid == (pytest.approx(-9.45), pytest.approx(5.6))
+
+ assert result.place_id == 554
+ assert result.parent_place_id == 152
+ assert result.linked_place_id is None
+ assert result.osm_object is None
+ assert result.admin_level == 15
+
+ assert result.names == {'ref': '34 425'}
+ assert result.address is None
+ assert result.extratags is None
+
+ assert result.housenumber is None
+ assert result.postcode is None
+ assert result.wikipedia is None
+
+ assert result.rank_search == 20
+ assert result.rank_address == 22
+ assert result.importance is None
+
+ assert result.country_code == 'gb'
+ assert result.indexed_date == import_date.replace(tzinfo=dt.timezone.utc)
+
+ assert result.address_rows is None
+ assert result.linked_rows is None
+ assert result.parented_rows is None
+ assert result.name_keywords is None
+ assert result.address_keywords is None
+
+ assert result.geometry == {'type': 'ST_Point'}
+
+
+def test_lookup_postcode_with_address_details(apiobj):
+ apiobj.add_postcode(place_id=9000,
+ parent_place_id=332,
+ postcode='34 425',
+ country_code='gb',
+ rank_search=25, rank_address=25)
+ apiobj.add_placex(place_id=332, osm_type='N', osm_id=3333,
+ class_='place', type='suburb', name='Smallplace',
+ country_code='gb', admin_level=13,
+ rank_search=24, rank_address=23)
+ apiobj.add_address_placex(332, fromarea=True, isaddress=True,
+ place_id=1001, osm_type='N', osm_id=3334,
+ class_='place', type='city', name='Bigplace',
+ country_code='gb',
+ rank_search=17, rank_address=16)
+
+ result = apiobj.api.lookup(napi.PlaceID(9000),
+ napi.LookupDetails(address_details=True))
+
+ assert result.address_rows == [
+ napi.AddressLine(place_id=332, osm_object=('N', 3333),
+ category=('place', 'suburb'),
+ names={'name': 'Smallplace'}, extratags={},
+ admin_level=13, fromarea=True, isaddress=True,
+ rank_address=23, distance=0.0),
+ napi.AddressLine(place_id=1001, osm_object=('N', 3334),
+ category=('place', 'city'),
+ names={'name': 'Bigplace'}, extratags={},
+ admin_level=15, fromarea=True, isaddress=True,
+ rank_address=16, distance=0.0),
+ napi.AddressLine(place_id=None, osm_object=None,
+ category=('place', 'postcode'),
+ names={'ref': '34 425'}, extratags={},
+ admin_level=None, fromarea=False, isaddress=True,
+ rank_address=5, distance=0.0),
+ napi.AddressLine(place_id=None, osm_object=None,
+ category=('place', 'country_code'),
+ names={'ref': 'gb'}, extratags={},
+ admin_level=None, fromarea=True, isaddress=False,
+ rank_address=4, distance=0.0)
+ ]
+
+@pytest.mark.parametrize('objid', [napi.PlaceID(1736),
+ napi.OsmID('W', 55),
+ napi.OsmID('N', 55, 'amenity')])
+def test_lookup_missing_object(apiobj, objid):
+ apiobj.add_placex(place_id=1, osm_type='N', osm_id=55,
+ class_='place', type='suburb')
+
+ assert apiobj.api.lookup(objid, napi.LookupDetails()) is None
+
+
+@pytest.mark.parametrize('gtype', (napi.GeometryFormat.KML,
+ napi.GeometryFormat.SVG,
+ napi.GeometryFormat.TEXT))
+def test_lookup_unsupported_geometry(apiobj, gtype):
+ apiobj.add_placex(place_id=332)
+
+ with pytest.raises(ValueError):
+ apiobj.api.lookup(napi.PlaceID(332),
+ napi.LookupDetails(geometry_output=gtype))
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2022 by the Nominatim developer community.
+# Copyright (C) 2023 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the status API call.
import pytest
from nominatim.version import NOMINATIM_VERSION, NominatimVersion
-from nominatim.api import NominatimAPI
+import nominatim.api as napi
-def test_status_no_extra_info(apiobj, table_factory):
- table_factory('import_status',
- definition="lastimportdate timestamp with time zone NOT NULL")
- table_factory('nominatim_properties',
- definition='property TEXT, value TEXT')
-
- result = apiobj.status()
+def test_status_no_extra_info(apiobj):
+ result = apiobj.api.status()
assert result.status == 0
assert result.message == 'OK'
assert result.data_updated is None
-def test_status_full(apiobj, table_factory):
- table_factory('import_status',
- definition="lastimportdate timestamp with time zone NOT NULL",
- content=(('2022-12-07 15:14:46+01',),))
- table_factory('nominatim_properties',
- definition='property TEXT, value TEXT',
- content=(('database_version', '99.5.4-2'), ))
+def test_status_full(apiobj):
+ import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0, tzinfo=dt.timezone.utc)
+ apiobj.add_data('import_status',
+ [{'lastimportdate': import_date}])
+ apiobj.add_data('properties',
+ [{'property': 'database_version', 'value': '99.5.4-2'}])
- result = apiobj.status()
+ result = apiobj.api.status()
assert result.status == 0
assert result.message == 'OK'
assert result.software_version == NOMINATIM_VERSION
assert result.database_version == NominatimVersion(99, 5, 4, 2)
- assert result.data_updated == dt.datetime(2022, 12, 7, 14, 14, 46, 0, tzinfo=dt.timezone.utc)
+ assert result.data_updated == import_date
def test_status_database_not_found(monkeypatch):
monkeypatch.setenv('NOMINATIM_DATABASE_DSN', 'dbname=rgjdfkgjedkrgdfkngdfkg')
- api = NominatimAPI(Path('/invalid'), {})
+ api = napi.NominatimAPI(Path('/invalid'), {})
result = api.status()
--- /dev/null
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# 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.
+"""
+Test functions for adapting results to the user's locale.
+"""
+import pytest
+
+from nominatim.api import Locales
+
+def test_display_name_empty_names():
+ l = Locales(['en', 'de'])
+
+ assert l.display_name(None) == ''
+ assert l.display_name({}) == ''
+
+def test_display_name_none_localized():
+ l = Locales()
+
+ assert l.display_name({}) == ''
+ assert l.display_name({'name:de': 'DE', 'name': 'ALL'}) == 'ALL'
+ assert l.display_name({'ref': '34', 'name:de': 'DE'}) == '34'
+
+
+def test_display_name_localized():
+ l = Locales(['en', 'de'])
+
+ assert l.display_name({}) == ''
+ assert l.display_name({'name:de': 'DE', 'name': 'ALL'}) == 'DE'
+ assert l.display_name({'ref': '34', 'name:de': 'DE'}) == 'DE'
+
+
+def test_display_name_preference():
+ l = Locales(['en', 'de'])
+
+ assert l.display_name({}) == ''
+ assert l.display_name({'name:de': 'DE', 'name:en': 'EN'}) == 'EN'
+ assert l.display_name({'official_name:en': 'EN', 'name:de': 'DE'}) == 'DE'
+
+
+@pytest.mark.parametrize('langstr,langlist',
+ [('fr', ['fr']),
+ ('fr-FR', ['fr-FR', 'fr']),
+ ('de,fr-FR', ['de', 'fr-FR', 'fr']),
+ ('fr,de,fr-FR', ['fr', 'de', 'fr-FR']),
+ ('en;q=0.5,fr', ['fr', 'en']),
+ ('en;q=0.5,fr,en-US', ['fr', 'en-US', 'en']),
+ ('en,fr;garbage,de', ['en', 'de'])])
+def test_from_language_preferences(langstr, langlist):
+ assert Locales.from_accept_languages(langstr).languages == langlist
Tests for formatting results for the V1 API.
"""
import datetime as dt
+import json
+
import pytest
import nominatim.api.v1 as api_impl
-from nominatim.api import StatusResult
+import nominatim.api as napi
from nominatim.version import NOMINATIM_VERSION
STATUS_FORMATS = {'text', 'json'}
# StatusResult
def test_status_format_list():
- assert set(api_impl.list_formats(StatusResult)) == STATUS_FORMATS
+ assert set(api_impl.list_formats(napi.StatusResult)) == STATUS_FORMATS
@pytest.mark.parametrize('fmt', list(STATUS_FORMATS))
def test_status_supported(fmt):
- assert api_impl.supports_format(StatusResult, fmt)
+ assert api_impl.supports_format(napi.StatusResult, fmt)
def test_status_unsupported():
- assert not api_impl.supports_format(StatusResult, 'gagaga')
+ assert not api_impl.supports_format(napi.StatusResult, 'gagaga')
def test_status_format_text():
- assert api_impl.format_result(StatusResult(0, 'message here'), 'text') == 'OK'
+ assert api_impl.format_result(napi.StatusResult(0, 'message here'), 'text', {}) == 'OK'
def test_status_format_text():
- assert api_impl.format_result(StatusResult(500, 'message here'), 'text') == 'ERROR: message here'
+ assert api_impl.format_result(napi.StatusResult(500, 'message here'), 'text', {}) == 'ERROR: message here'
def test_status_format_json_minimal():
- status = StatusResult(700, 'Bad format.')
+ status = napi.StatusResult(700, 'Bad format.')
- result = api_impl.format_result(status, 'json')
+ result = api_impl.format_result(status, 'json', {})
assert result == '{"status":700,"message":"Bad format.","software_version":"%s"}' % (NOMINATIM_VERSION, )
def test_status_format_json_full():
- status = StatusResult(0, 'OK')
+ status = napi.StatusResult(0, 'OK')
status.data_updated = dt.datetime(2010, 2, 7, 20, 20, 3, 0, tzinfo=dt.timezone.utc)
status.database_version = '5.6'
- result = api_impl.format_result(status, 'json')
+ result = api_impl.format_result(status, 'json', {})
assert result == '{"status":0,"message":"OK","data_updated":"2010-02-07T20:20:03+00:00","software_version":"%s","database_version":"5.6"}' % (NOMINATIM_VERSION, )
+
+
+# SearchResult
+
+def test_search_details_minimal():
+ search = napi.SearchResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0))
+
+ result = api_impl.format_result(search, 'details-json', {})
+
+ assert json.loads(result) == \
+ {'category': 'place',
+ 'type': 'thing',
+ 'admin_level': 15,
+ 'localname': '',
+ 'calculated_importance': pytest.approx(0.0000001),
+ 'rank_address': 30,
+ 'rank_search': 30,
+ 'isarea': False,
+ 'centroid': {'type': 'Point', 'coordinates': [1.0, 2.0]},
+ 'geometry': {'type': 'Point', 'coordinates': [1.0, 2.0]},
+ }
+
+
+def test_search_details_full():
+ import_date = dt.datetime(2010, 2, 7, 20, 20, 3, 0)
+ search = napi.SearchResult(
+ source_table=napi.SourceTable.PLACEX,
+ category=('amenity', 'bank'),
+ centroid=napi.Point(56.947, -87.44),
+ place_id=37563,
+ parent_place_id=114,
+ linked_place_id=55693,
+ osm_object=('W', 442100),
+ admin_level=14,
+ names={'name': 'Bank', 'name:fr': 'Banque'},
+ address={'city': 'Niento', 'housenumber': ' 3'},
+ extratags={'atm': 'yes'},
+ housenumber='3',
+ postcode='556 X23',
+ wikipedia='en:Bank',
+ rank_address=29,
+ rank_search=28,
+ importance=0.0443,
+ country_code='ll',
+ indexed_date = import_date
+ )
+
+ result = api_impl.format_result(search, 'details-json', {})
+
+ assert json.loads(result) == \
+ {'place_id': 37563,
+ 'parent_place_id': 114,
+ 'osm_type': 'W',
+ 'osm_id': 442100,
+ 'category': 'amenity',
+ 'type': 'bank',
+ 'admin_level': 14,
+ 'localname': 'Bank',
+ 'names': {'name': 'Bank', 'name:fr': 'Banque'},
+ 'addresstags': {'city': 'Niento', 'housenumber': ' 3'},
+ 'housenumber': '3',
+ 'calculated_postcode': '556 X23',
+ 'country_code': 'll',
+ 'indexed_date': '2010-02-07T20:20:03+00:00',
+ 'importance': pytest.approx(0.0443),
+ 'calculated_importance': pytest.approx(0.0443),
+ 'extratags': {'atm': 'yes'},
+ 'calculated_wikipedia': 'en:Bank',
+ 'rank_address': 29,
+ 'rank_search': 28,
+ 'isarea': False,
+ 'centroid': {'type': 'Point', 'coordinates': [56.947, -87.44]},
+ 'geometry': {'type': 'Point', 'coordinates': [56.947, -87.44]},
+ }
+
+
+@pytest.mark.parametrize('gtype,isarea', [('ST_Point', False),
+ ('ST_LineString', False),
+ ('ST_Polygon', True),
+ ('ST_MultiPolygon', True)])
+def test_search_details_no_geometry(gtype, isarea):
+ search = napi.SearchResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0),
+ geometry={'type': gtype})
+
+ result = api_impl.format_result(search, 'details-json', {})
+ js = json.loads(result)
+
+ assert js['geometry'] == {'type': 'Point', 'coordinates': [1.0, 2.0]}
+ assert js['isarea'] == isarea
+
+
+def test_search_details_with_geometry():
+ search = napi.SearchResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0),
+ geometry={'geojson': '{"type":"Point","coordinates":[56.947,-87.44]}'})
+
+ result = api_impl.format_result(search, 'details-json', {})
+ js = json.loads(result)
+
+ assert js['geometry'] == {'type': 'Point', 'coordinates': [56.947, -87.44]}
+ assert js['isarea'] == False
+
+
+def test_search_details_with_address_minimal():
+ search = napi.SearchResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0),
+ address_rows=[
+ napi.AddressLine(place_id=None,
+ osm_object=None,
+ category=('bnd', 'note'),
+ names={},
+ extratags=None,
+ admin_level=None,
+ fromarea=False,
+ isaddress=False,
+ rank_address=10,
+ distance=0.0)
+ ])
+
+ result = api_impl.format_result(search, 'details-json', {})
+ js = json.loads(result)
+
+ assert js['address'] == [{'localname': '',
+ 'class': 'bnd',
+ 'type': 'note',
+ 'rank_address': 10,
+ 'distance': 0.0,
+ 'isaddress': False}]
+
+
+def test_search_details_with_address_full():
+ search = napi.SearchResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0),
+ address_rows=[
+ napi.AddressLine(place_id=3498,
+ osm_object=('R', 442),
+ category=('bnd', 'note'),
+ names={'name': 'Trespass'},
+ extratags={'access': 'no',
+ 'place_type': 'spec'},
+ admin_level=4,
+ fromarea=True,
+ isaddress=True,
+ rank_address=10,
+ distance=0.034)
+ ])
+
+ result = api_impl.format_result(search, 'details-json', {})
+ js = json.loads(result)
+
+ assert js['address'] == [{'localname': 'Trespass',
+ 'place_id': 3498,
+ 'osm_id': 442,
+ 'osm_type': 'R',
+ 'place_type': 'spec',
+ 'class': 'bnd',
+ 'type': 'note',
+ 'admin_level': 4,
+ 'rank_address': 10,
+ 'distance': 0.034,
+ 'isaddress': True}]
@pytest.mark.parametrize("params", [('search', '--query', 'new'),
('search', '--city', 'Berlin'),
('reverse', '--lat', '0', '--lon', '0', '--zoom', '13'),
- ('lookup', '--id', 'N1'),
- ('details', '--node', '1'),
- ('details', '--way', '1'),
- ('details', '--relation', '1'),
- ('details', '--place_id', '10001')])
+ ('lookup', '--id', 'N1')])
class TestCliApiCallPhp:
@pytest.fixture(autouse=True)
json.loads(capsys.readouterr().out)
+class TestCliDetailsCall:
+
+ @pytest.fixture(autouse=True)
+ def setup_status_mock(self, monkeypatch):
+ result = napi.SearchResult(napi.SourceTable.PLACEX, ('place', 'thing'),
+ (1.0, -3.0))
+
+ monkeypatch.setattr(napi.NominatimAPI, 'lookup',
+ lambda *args: result)
+
+ @pytest.mark.parametrize("params", [('--node', '1'),
+ ('--way', '1'),
+ ('--relation', '1'),
+ ('--place_id', '10001')])
+
+ def test_status_json_format(self, cli_call, tmp_path, capsys, params):
+ result = cli_call('details', '--project-dir', str(tmp_path), *params)
+
+ assert result == 0
+
+ json.loads(capsys.readouterr().out)
+
+
QUERY_PARAMS = {
'search': ('--query', 'somewhere'),
'reverse': ('--lat', '20', '--lon', '30'),
assert cli_call('search', *QUERY_PARAMS['search'], '--project-dir', str(project_env.project_dir),
'--no-dedupe') == 0
-
-
-def test_cli_details_param_class(cli_call, project_env):
- webdir = project_env.project_dir / 'website'
- webdir.mkdir()
- (webdir / 'details.php').write_text(f"""<?php
- exit($_GET['class'] == 'highway' ? 0 : 10);
- """)
-
- assert cli_call('details', *QUERY_PARAMS['details'], '--project-dir', str(project_env.project_dir),
- '--class', 'highway') == 0
-
-
-@pytest.mark.parametrize('param', ('lang', 'accept-language'))
-def test_cli_details_param_lang(cli_call, project_env, param):
- webdir = project_env.project_dir / 'website'
- webdir.mkdir()
- (webdir / 'details.php').write_text(f"""<?php
- exit($_GET['accept-language'] == 'es' ? 0 : 10);
- """)
-
- assert cli_call('details', *QUERY_PARAMS['details'], '--project-dir', str(project_env.project_dir),
- '--' + param, 'es') == 0
-
projects / nominatim.git / commitdiff