+async def _unstructured_search(query: str, api: napi.NominatimAPIAsync,
+ details: Dict[str, Any]) -> napi.SearchResults:
+ if not query:
+ return napi.SearchResults()
+
+ # Extract special format for coordinates from query.
+ query, x, y = helpers.extract_coords_from_query(query)
+ if x is not None:
+ assert y is not None
+ details['near'] = napi.Point(x, y)
+ details['near_radius'] = 0.1
+
+ # If no query is left, revert to reverse search.
+ if x is not None and not query:
+ result = await api.reverse(details['near'], **details)
+ if not result:
+ return napi.SearchResults()
+
+ return napi.SearchResults(
+ [napi.SearchResult(**{f.name: getattr(result, f.name)
+ for f in dataclasses.fields(napi.SearchResult)
+ if hasattr(result, f.name)})])
+
+ query, cls, typ = helpers.extract_category_from_query(query)
+ if cls is not None:
+ assert typ is not None
+ return await api.search_category([(cls, typ)], near_query=query, **details)
+
+ return await api.search(query, **details)
+
+
+async def search_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
+ """ Server glue for /search endpoint. See API docs for details.
+ """
+ fmt = params.parse_format(napi.SearchResults, 'jsonv2')
+ debug = params.setup_debugging()
+ details = params.parse_geometry_details(fmt)
+
+ details['countries'] = params.get('countrycodes', None)
+ details['excluded'] = params.get('exclude_place_ids', None)
+ details['viewbox'] = params.get('viewbox', None) or params.get('viewboxlbrt', None)
+ details['bounded_viewbox'] = params.get_bool('bounded', False)
+ details['dedupe'] = params.get_bool('dedupe', True)
+
+ max_results = max(1, min(50, params.get_int('limit', 10)))
+ details['max_results'] = max_results + min(10, max_results) \
+ if details['dedupe'] else max_results
+
+ details['min_rank'], details['max_rank'] = \
+ helpers.feature_type_to_rank(params.get('featureType', ''))
+ if params.get('featureType', None) is not None:
+ details['layers'] = napi.DataLayer.ADDRESS
+ else:
+ details['layers'] = params.get_layers()
+
+ details['locales'] = napi.Locales.from_accept_languages(params.get_accepted_languages())
+
+ # unstructured query parameters
+ query = params.get('q', None)
+ # structured query parameters
+ queryparts = {}
+ for key in ('amenity', 'street', 'city', 'county', 'state', 'postalcode', 'country'):
+ details[key] = params.get(key, None)
+ if details[key]:
+ queryparts[key] = details[key]
+
+ try:
+ if query is not None:
+ if queryparts:
+ params.raise_error("Structured query parameters"
+ "(amenity, street, city, county, state, postalcode, country)"
+ " cannot be used together with 'q' parameter.")
+ queryparts['q'] = query
+ results = await _unstructured_search(query, api, details)
+ else:
+ query = ', '.join(queryparts.values())
+
+ results = await api.search_address(**details)
+ except UsageError as err:
+ params.raise_error(str(err))
+
+ if details['dedupe'] and len(results) > 1:
+ results = helpers.deduplicate_results(results, max_results)
+
+ if debug:
+ return params.build_response(loglib.get_and_disable(), num_results=len(results))
+
+ if fmt == 'xml':
+ helpers.extend_query_parts(queryparts, details,
+ params.get('featureType', ''),
+ params.get_bool('namedetails', False),
+ params.get_bool('extratags', False),
+ (str(r.place_id) for r in results if r.place_id))
+ queryparts['format'] = fmt
+
+ moreurl = params.base_uri() + '/search?' + urlencode(queryparts)
+ else:
+ moreurl = ''
+
+ fmt_options = {'query': query, 'more_url': moreurl,
+ 'exclude_place_ids': queryparts.get('exclude_place_ids'),
+ 'viewbox': queryparts.get('viewbox'),
+ 'extratags': params.get_bool('extratags', False),
+ 'namedetails': params.get_bool('namedetails', False),
+ 'addressdetails': params.get_bool('addressdetails', False)}
+
+ output = formatting.format_result(results, fmt, fmt_options)
+
+ return params.build_response(output, num_results=len(results))
+
+
+async def deletable_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
+ """ Server glue for /deletable endpoint.
+ This is a special endpoint that shows polygons that have been
+ deleted or are broken in the OSM data but are kept in the
+ Nominatim database to minimize disruption.
+ """
+ fmt = params.parse_format(RawDataList, 'json')
+
+ async with api.begin() as conn:
+ sql = sa.text(""" SELECT p.place_id, country_code,
+ name->'name' as name, i.*
+ FROM placex p, import_polygon_delete i
+ WHERE p.osm_id = i.osm_id AND p.osm_type = i.osm_type
+ AND p.class = i.class AND p.type = i.type
+ """)
+ results = RawDataList(r._asdict() for r in await conn.execute(sql))
+
+ return params.build_response(formatting.format_result(results, fmt, {}))
+
+
+async def polygons_endpoint(api: napi.NominatimAPIAsync, params: ASGIAdaptor) -> Any:
+ """ Server glue for /polygons endpoint.
+ This is a special endpoint that shows polygons that have changed
+ thier size but are kept in the Nominatim database with their
+ old area to minimize disruption.
+ """
+ fmt = params.parse_format(RawDataList, 'json')
+ sql_params: Dict[str, Any] = {
+ 'days': params.get_int('days', -1),
+ 'cls': params.get('class')
+ }
+ reduced = params.get_bool('reduced', False)
+
+ async with api.begin() as conn:
+ sql = sa.select(sa.text("""osm_type, osm_id, class, type,
+ name->'name' as name,
+ country_code, errormessage, updated"""))\
+ .select_from(sa.text('import_polygon_error'))
+ if sql_params['days'] > 0:
+ sql = sql.where(sa.text("updated > 'now'::timestamp - make_interval(days => :days)"))
+ if reduced:
+ sql = sql.where(sa.text("errormessage like 'Area reduced%'"))
+ if sql_params['cls'] is not None:
+ sql = sql.where(sa.text("class = :cls"))
+
+ sql = sql.order_by(sa.literal_column('updated').desc()).limit(1000)
+
+ results = RawDataList(r._asdict() for r in await conn.execute(sql, sql_params))
+
+ return params.build_response(formatting.format_result(results, fmt, {}))
+
+
+EndpointFunc = Callable[[napi.NominatimAPIAsync, ASGIAdaptor], Any]
projects / nominatim.git / blobdiff