introduce slim Geometry database type

nominatim/db/sqlalchemy_schema.py

@@ -10,10 +10,11 @@ SQLAlchemy definitions for all tables used by the frontend.
 from typing import Any
 
 import sqlalchemy as sa
-from geoalchemy2 import Geometry
 from sqlalchemy.dialects.postgresql import HSTORE, ARRAY, JSONB
 from sqlalchemy.dialects.sqlite import JSON as sqlite_json
 
+from nominatim.db.sqlalchemy_types import Geometry
+
 class PostgresTypes:
     """ Type definitions for complex types as used in Postgres variants.
     """
@@ -72,12 +73,12 @@ class SearchTables:
             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('geometry', Geometry, nullable=False),
             sa.Column('wikipedia', sa.Text),
             sa.Column('country_code', sa.String(2)),
             sa.Column('housenumber', sa.Text),
             sa.Column('postcode', sa.Text),
-            sa.Column('centroid', Geometry(srid=4326, spatial_index=False)))
+            sa.Column('centroid', Geometry))
 
         self.addressline = sa.Table('place_addressline', meta,
             sa.Column('place_id', sa.BigInteger, index=True),
@@ -96,7 +97,7 @@ class SearchTables:
             sa.Column('indexed_date', sa.DateTime),
             sa.Column('country_code', sa.String(2)),
             sa.Column('postcode', sa.Text, index=True),
-            sa.Column('geometry', Geometry(srid=4326)))
+            sa.Column('geometry', Geometry))
 
         self.osmline = sa.Table('location_property_osmline', meta,
             sa.Column('place_id', sa.BigInteger, nullable=False, unique=True),
@@ -108,7 +109,7 @@ class SearchTables:
             sa.Column('step', sa.SmallInteger),
             sa.Column('partition', sa.SmallInteger),
             sa.Column('indexed_status', sa.SmallInteger),
-            sa.Column('linegeo', Geometry(srid=4326)),
+            sa.Column('linegeo', Geometry),
             sa.Column('address', self.types.Composite),
             sa.Column('postcode', sa.Text),
             sa.Column('country_code', sa.String(2)))
@@ -123,7 +124,7 @@ class SearchTables:
         self.country_grid = sa.Table('country_osm_grid', meta,
             sa.Column('country_code', sa.String(2)),
             sa.Column('area', sa.Float),
-            sa.Column('geometry', Geometry(srid=4326)))
+            sa.Column('geometry', Geometry))
 
         # The following tables are not necessarily present.
         self.search_name = sa.Table('search_name', meta,
@@ -134,7 +135,7 @@ class SearchTables:
             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)))
+            sa.Column('centroid', Geometry))
 
         self.tiger = sa.Table('location_property_tiger', meta,
             sa.Column('place_id', sa.BigInteger),
@@ -143,5 +144,5 @@ class SearchTables:
             sa.Column('endnumber', sa.Integer),
             sa.Column('step', sa.SmallInteger),
             sa.Column('partition', sa.SmallInteger),
-            sa.Column('linegeo', Geometry(srid=4326, spatial_index=False)),
+            sa.Column('linegeo', Geometry),
             sa.Column('postcode', sa.Text))

nominatim/db/sqlalchemy_types.py

@@ -0,0 +1,93 @@
+# 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.
+"""
+Custom types for SQLAlchemy.
+"""
+from typing import Callable, Any
+
+import sqlalchemy as sa
+import sqlalchemy.types as types
+
+from nominatim.typing import SaColumn
+
+class Geometry(types.UserDefinedType[Any]):
+    """ Simplified type decorator for PostGIS geometry. This type
+        only supports geometries in 4326 projection.
+    """
+    cache_ok = True
+
+    def __init__(self, subtype: str = 'Geometry'):
+        self.subtype = subtype
+
+
+    def get_col_spec(self) -> str:
+        return f'GEOMETRY({self.subtype}, 4326)'
+
+
+    def bind_processor(self, dialect: sa.Dialect) -> Callable[[Any], str]:
+        def process(value: Any) -> str:
+            assert isinstance(value, str)
+            return value
+        return process
+
+
+    def result_processor(self, dialect: sa.Dialect, coltype: object) -> Callable[[Any], str]:
+        def process(value: Any) -> str:
+            assert isinstance(value, str)
+            return value
+        return process
+
+
+    def bind_expression(self, bindvalue: sa.BindParameter[Any]) -> SaColumn:
+        return sa.func.ST_GeomFromText(bindvalue, type_=self)
+
+
+    class comparator_factory(types.UserDefinedType.Comparator):
+
+        def is_line_like(self) -> SaColumn:
+            return sa.func.ST_GeometryType(self, type_=sa.String).in_(('ST_LineString',
+                                                                       'ST_MultiLineString'))
+
+        def is_area(self) -> SaColumn:
+            return sa.func.ST_GeometryType(self, type_=sa.String).in_(('ST_Polygon',
+                                                                       'ST_MultiPolygon'))
+
+
+        def ST_DWithin(self, other: SaColumn, distance: SaColumn) -> SaColumn:
+            return sa.func.ST_DWithin(self, other, distance, type_=sa.Float)
+
+
+        def ST_Distance(self, other: SaColumn) -> SaColumn:
+            return sa.func.ST_Distance(self, other, type_=sa.Float)
+
+
+        def ST_Contains(self, other: SaColumn) -> SaColumn:
+            return sa.func.ST_Contains(self, other, type_=sa.Float)
+
+
+        def ST_ClosestPoint(self, other: SaColumn) -> SaColumn:
+            return sa.func.ST_ClosestPoint(self, other, type_=Geometry)
+
+
+        def ST_Buffer(self, other: SaColumn) -> SaColumn:
+            return sa.func.ST_Buffer(self, other, type_=Geometry)
+
+
+        def ST_Expand(self, other: SaColumn) -> SaColumn:
+            return sa.func.ST_Expand(self, other, type_=Geometry)
+
+
+        def ST_Centroid(self) -> SaColumn:
+            return sa.func.ST_Centroid(self, type_=Geometry)
+
+
+        def ST_LineInterpolatePoint(self, other: SaColumn) -> SaColumn:
+            return sa.func.ST_LineInterpolatePoint(self, other, type_=Geometry)
+
+
+        def ST_LineLocatePoint(self, other: SaColumn) -> SaColumn:
+            return sa.func.ST_LineLocatePoint(self, other, type_=sa.Float)