import sqlalchemy as sa
from sqlalchemy.ext.compiler import compiles
-from nominatim_core.typing import SaColumn
+from ..typing import SaColumn
-# pylint: disable=all
class PlacexGeometryReverseLookuppolygon(sa.sql.functions.GenericFunction[Any]):
""" Check for conditions that allow partial index use on
inherit_cache = True
-@compiles(PlacexGeometryReverseLookuppolygon) # type: ignore[no-untyped-call, misc]
+@compiles(PlacexGeometryReverseLookuppolygon)
def _default_intersects(element: PlacexGeometryReverseLookuppolygon,
compiler: 'sa.Compiled', **kw: Any) -> str:
return ("(ST_GeometryType(placex.geometry) in ('ST_Polygon', 'ST_MultiPolygon')"
" AND placex.linked_place_id is null)")
-@compiles(PlacexGeometryReverseLookuppolygon, 'sqlite') # type: ignore[no-untyped-call, misc]
+@compiles(PlacexGeometryReverseLookuppolygon, 'sqlite')
def _sqlite_intersects(element: PlacexGeometryReverseLookuppolygon,
compiler: 'sa.Compiled', **kw: Any) -> str:
return ("(ST_GeometryType(placex.geometry) in ('POLYGON', 'MULTIPOLYGON')"
self.tablename = table.name
-@compiles(IntersectsReverseDistance) # type: ignore[no-untyped-call, misc]
+@compiles(IntersectsReverseDistance)
def default_reverse_place_diameter(element: IntersectsReverseDistance,
compiler: 'sa.Compiled', **kw: Any) -> str:
table = element.tablename
f" AND {table}.name is not null"\
f" AND {table}.linked_place_id is null"\
f" AND {table}.osm_type = 'N'" + \
- " AND ST_Buffer(%s, reverse_place_diameter(%s)) && %s)" % \
- tuple(map(lambda c: compiler.process(c, **kw), element.clauses))
+ " AND ST_Buffer(%s, reverse_place_diameter(%s)) && %s)" \
+ % tuple(map(lambda c: compiler.process(c, **kw), element.clauses))
-@compiles(IntersectsReverseDistance, 'sqlite') # type: ignore[no-untyped-call, misc]
+@compiles(IntersectsReverseDistance, 'sqlite')
def sqlite_reverse_place_diameter(element: IntersectsReverseDistance,
compiler: 'sa.Compiled', **kw: Any) -> str:
geom1, rank, geom2 = list(element.clauses)
table = element.tablename
- return (f"({table}.rank_address between 4 and 25"\
- f" AND {table}.type != 'postcode'"\
- f" AND {table}.name is not null"\
- f" AND {table}.linked_place_id is null"\
- f" AND {table}.osm_type = 'N'"\
- " AND MbrIntersects(%s, ST_Expand(%s, 14.0 * exp(-0.2 * %s) - 0.03))"\
- f" AND {table}.place_id IN"\
- " (SELECT place_id FROM placex_place_node_areas"\
- " WHERE ROWID IN (SELECT ROWID FROM SpatialIndex"\
- " WHERE f_table_name = 'placex_place_node_areas'"\
- " AND search_frame = %s)))") % (
+ return (f"({table}.rank_address between 4 and 25"
+ f" AND {table}.type != 'postcode'"
+ f" AND {table}.name is not null"
+ f" AND {table}.linked_place_id is null"
+ f" AND {table}.osm_type = 'N'"
+ " AND MbrIntersects(%s, ST_Expand(%s, 14.0 * exp(-0.2 * %s) - 0.03))"
+ f" AND {table}.place_id IN"
+ " (SELECT place_id FROM placex_place_node_areas"
+ " WHERE ROWID IN (SELECT ROWID FROM SpatialIndex"
+ " WHERE f_table_name = 'placex_place_node_areas'"
+ " AND search_frame = %s)))") % (
compiler.process(geom1, **kw),
compiler.process(geom2, **kw),
compiler.process(rank, **kw),
inherit_cache = True
-@compiles(IsBelowReverseDistance) # type: ignore[no-untyped-call, misc]
+@compiles(IsBelowReverseDistance)
def default_is_below_reverse_distance(element: IsBelowReverseDistance,
compiler: 'sa.Compiled', **kw: Any) -> str:
dist, rank = list(element.clauses)
compiler.process(rank, **kw))
-@compiles(IsBelowReverseDistance, 'sqlite') # type: ignore[no-untyped-call, misc]
+@compiles(IsBelowReverseDistance, 'sqlite')
def sqlite_is_below_reverse_distance(element: IsBelowReverseDistance,
compiler: 'sa.Compiled', **kw: Any) -> str:
dist, rank = list(element.clauses)
table.c.housenumber, table.c.name)
-@compiles(IsAddressPoint) # type: ignore[no-untyped-call, misc]
+@compiles(IsAddressPoint)
def default_is_address_point(element: IsAddressPoint,
compiler: 'sa.Compiled', **kw: Any) -> str:
rank, hnr, name = list(element.clauses)
compiler.process(name, **kw))
-@compiles(IsAddressPoint, 'sqlite') # type: ignore[no-untyped-call, misc]
+@compiles(IsAddressPoint, 'sqlite')
def sqlite_is_address_point(element: IsAddressPoint,
compiler: 'sa.Compiled', **kw: Any) -> str:
rank, hnr, name = list(element.clauses)
name = 'CrosscheckNames'
inherit_cache = True
-@compiles(CrosscheckNames) # type: ignore[no-untyped-call, misc]
+
+@compiles(CrosscheckNames)
def compile_crosscheck_names(element: CrosscheckNames,
compiler: 'sa.Compiled', **kw: Any) -> str:
arg1, arg2 = list(element.clauses)
compiler.process(arg1, **kw), compiler.process(arg2, **kw))
-@compiles(CrosscheckNames, 'sqlite') # type: ignore[no-untyped-call, misc]
+@compiles(CrosscheckNames, 'sqlite')
def compile_sqlite_crosscheck_names(element: CrosscheckNames,
compiler: 'sa.Compiled', **kw: Any) -> str:
arg1, arg2 = list(element.clauses)
inherit_cache = True
-@compiles(JsonArrayEach) # type: ignore[no-untyped-call, misc]
+@compiles(JsonArrayEach)
def default_json_array_each(element: JsonArrayEach, compiler: 'sa.Compiled', **kw: Any) -> str:
return "json_array_elements(%s)" % compiler.process(element.clauses, **kw)
-@compiles(JsonArrayEach, 'sqlite') # type: ignore[no-untyped-call, misc]
+@compiles(JsonArrayEach, 'sqlite')
def sqlite_json_array_each(element: JsonArrayEach, compiler: 'sa.Compiled', **kw: Any) -> str:
return "json_each(%s)" % compiler.process(element.clauses, **kw)
-
class Greatest(sa.sql.functions.GenericFunction[Any]):
""" Function to compute maximum of all its input parameters.
"""
inherit_cache = True
-@compiles(Greatest, 'sqlite') # type: ignore[no-untyped-call, misc]
+@compiles(Greatest, 'sqlite')
def sqlite_greatest(element: Greatest, compiler: 'sa.Compiled', **kw: Any) -> str:
return "max(%s)" % compiler.process(element.clauses, **kw)
-
class RegexpWord(sa.sql.functions.GenericFunction[Any]):
""" Check if a full word is in a given string.
"""
inherit_cache = True
-@compiles(RegexpWord, 'postgresql') # type: ignore[no-untyped-call, misc]
+@compiles(RegexpWord, 'postgresql')
def postgres_regexp_nocase(element: RegexpWord, compiler: 'sa.Compiled', **kw: Any) -> str:
arg1, arg2 = list(element.clauses)
- return "%s ~* ('\\m(' || %s || ')\\M')::text" % (compiler.process(arg2, **kw), compiler.process(arg1, **kw))
+ return "%s ~* ('\\m(' || %s || ')\\M')::text" \
+ % (compiler.process(arg2, **kw), compiler.process(arg1, **kw))
-@compiles(RegexpWord, 'sqlite') # type: ignore[no-untyped-call, misc]
+@compiles(RegexpWord, 'sqlite')
def sqlite_regexp_nocase(element: RegexpWord, compiler: 'sa.Compiled', **kw: Any) -> str:
arg1, arg2 = list(element.clauses)
- return "regexp('\\b(' || %s || ')\\b', %s)" % (compiler.process(arg1, **kw), compiler.process(arg2, **kw))
+ return "regexp('\\b(' || %s || ')\\b', %s)"\
+ % (compiler.process(arg1, **kw), compiler.process(arg2, **kw))
projects / nominatim.git / blobdiff