nominatim/db/sqlalchemy_functions.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 Custom functions and expressions for SQLAlchemy.
   9 """
  10 from typing import Any
  11
  12 import sqlalchemy as sa
  13 from sqlalchemy.sql.expression import FunctionElement
  14 from sqlalchemy.ext.compiler import compiles
  15
  16 from nominatim.typing import SaColumn
  17
  18 # pylint: disable=abstract-method,missing-function-docstring,consider-using-f-string
  19
  20 def select_index_placex_geometry_reverse_lookuppolygon(table: str) -> 'sa.TextClause':
  21     """ Create an expression with the necessary conditions over a placex
  22         table that the index 'idx_placex_geometry_reverse_lookupPolygon'
  23         can be used.
  24     """
  25     return sa.text(f"ST_GeometryType({table}.geometry) in ('ST_Polygon', 'ST_MultiPolygon')"
  26                    f" AND {table}.rank_address between 4 and 25"
  27                    f" AND {table}.type != 'postcode'"
  28                    f" AND {table}.name is not null"
  29                    f" AND {table}.indexed_status = 0"
  30                    f" AND {table}.linked_place_id is null")
  31
  32 def select_index_placex_geometry_reverse_lookupplacenode(table: str) -> 'sa.TextClause':
  33     """ Create an expression with the necessary conditions over a placex
  34         table that the index 'idx_placex_geometry_reverse_lookupPlaceNode'
  35         can be used.
  36     """
  37     return sa.text(f"{table}.rank_address between 4 and 25"
  38                    f" AND {table}.type != 'postcode'"
  39                    f" AND {table}.name is not null"
  40                    f" AND {table}.linked_place_id is null"
  41                    f" AND {table}.osm_type = 'N'")
  42
  43
  44 class CrosscheckNames(FunctionElement[Any]):
  45     """ Check if in the given list of names in parameters 1 any of the names
  46         from the JSON array in parameter 2 are contained.
  47     """
  48     name = 'CrosscheckNames'
  49     inherit_cache = True
  50
  51 @compiles(CrosscheckNames) # type: ignore[no-untyped-call, misc]
  52 def compile_crosscheck_names(element: SaColumn,
  53                              compiler: 'sa.Compiled', **kw: Any) -> str:
  54     arg1, arg2 = list(element.clauses)
  55     return "coalesce(avals(%s) && ARRAY(SELECT * FROM json_array_elements_text(%s)), false)" % (
  56             compiler.process(arg1, **kw), compiler.process(arg2, **kw))