nominatim/api/lookup.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2023 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Implementation of place lookup by ID.
   9 """
  10 from typing import Optional
  11
  12 import sqlalchemy as sa
  13
  14 from nominatim.typing import SaColumn, SaLabel, SaRow
  15 from nominatim.api.connection import SearchConnection
  16 import nominatim.api.types as ntyp
  17 import nominatim.api.results as nres
  18
  19 def _select_column_geometry(column: SaColumn,
  20                             geometry_output: ntyp.GeometryFormat) -> SaLabel:
  21     """ Create the appropriate column expression for selecting a
  22         geometry for the details response.
  23     """
  24     if geometry_output & ntyp.GeometryFormat.GEOJSON:
  25         return sa.literal_column(f"""
  26                   ST_AsGeoJSON(CASE WHEN ST_NPoints({0}) > 5000
  27                                THEN ST_SimplifyPreserveTopology({0}, 0.0001)
  28                                ELSE {column.name} END)
  29                   """).label('geometry_geojson')
  30
  31     return sa.func.ST_GeometryType(column).label('geometry_type')
  32
  33
  34 async def find_in_placex(conn: SearchConnection, place: ntyp.PlaceRef,
  35                          details: ntyp.LookupDetails) -> Optional[SaRow]:
  36     """ Search for the given place in the placex table and return the
  37         base information.
  38     """
  39     t = conn.t.placex
  40     sql = sa.select(t.c.place_id, t.c.osm_type, t.c.osm_id, t.c.name,
  41                     t.c.class_, t.c.type, t.c.admin_level,
  42                     t.c.address, t.c.extratags,
  43                     t.c.housenumber, t.c.postcode, t.c.country_code,
  44                     t.c.importance, t.c.wikipedia, t.c.indexed_date,
  45                     t.c.parent_place_id, t.c.rank_address, t.c.rank_search,
  46                     t.c.linked_place_id,
  47                     sa.func.ST_X(t.c.centroid).label('x'),
  48                     sa.func.ST_Y(t.c.centroid).label('y'),
  49                     _select_column_geometry(t.c.geometry, details.geometry_output))
  50
  51     if isinstance(place, ntyp.PlaceID):
  52         sql = sql.where(t.c.place_id == place.place_id)
  53     elif isinstance(place, ntyp.OsmID):
  54         sql = sql.where(t.c.osm_type == place.osm_type)\
  55                  .where(t.c.osm_id == place.osm_id)
  56         if place.osm_class:
  57             sql = sql.where(t.c.class_ == place.osm_class)
  58         else:
  59             sql = sql.order_by(t.c.class_)
  60         sql = sql.limit(1)
  61     else:
  62         return None
  63
  64     return (await conn.execute(sql)).one_or_none()
  65
  66
  67 async def get_place_by_id(conn: SearchConnection, place: ntyp.PlaceRef,
  68                           details: ntyp.LookupDetails) -> Optional[nres.SearchResult]:
  69     """ Retrieve a place with additional details from the database.
  70     """
  71     if details.geometry_output and details.geometry_output != ntyp.GeometryFormat.GEOJSON:
  72         raise ValueError("lookup only supports geojosn polygon output.")
  73
  74     row = await find_in_placex(conn, place, details)
  75     if row is not None:
  76         result = nres.create_from_placex_row(row=row)
  77         await nres.add_result_details(conn, result, details)
  78         return result
  79
  80     # Nothing found under this ID.
  81     return None