nominatim/db/sqlalchemy_types.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 types for SQLAlchemy.
   9 """
  10 from typing import Callable, Any
  11
  12 import sqlalchemy as sa
  13 from sqlalchemy import types
  14
  15 from nominatim.typing import SaColumn, SaBind
  16
  17 #pylint: disable=all
  18
  19 class Geometry(types.UserDefinedType[Any]):
  20     """ Simplified type decorator for PostGIS geometry. This type
  21         only supports geometries in 4326 projection.
  22     """
  23     cache_ok = True
  24
  25     def __init__(self, subtype: str = 'Geometry'):
  26         self.subtype = subtype
  27
  28
  29     def get_col_spec(self) -> str:
  30         return f'GEOMETRY({self.subtype}, 4326)'
  31
  32
  33     def bind_processor(self, dialect: 'sa.Dialect') -> Callable[[Any], str]:
  34         def process(value: Any) -> str:
  35             if isinstance(value, str):
  36                 return 'SRID=4326;' + value
  37
  38             return 'SRID=4326;' + value.to_wkt()
  39         return process
  40
  41
  42     def result_processor(self, dialect: 'sa.Dialect', coltype: object) -> Callable[[Any], str]:
  43         def process(value: Any) -> str:
  44             assert isinstance(value, str)
  45             return value
  46         return process
  47
  48
  49     def bind_expression(self, bindvalue: SaBind) -> SaColumn:
  50         return sa.func.ST_GeomFromText(bindvalue, type_=self)
  51
  52
  53     class comparator_factory(types.UserDefinedType.Comparator):
  54
  55         def intersects(self, other: SaColumn) -> 'sa.Operators':
  56             return self.op('&&')(other)
  57
  58         def is_line_like(self) -> SaColumn:
  59             return sa.func.ST_GeometryType(self, type_=sa.String).in_(('ST_LineString',
  60                                                                        'ST_MultiLineString'))
  61
  62         def is_area(self) -> SaColumn:
  63             return sa.func.ST_GeometryType(self, type_=sa.String).in_(('ST_Polygon',
  64                                                                        'ST_MultiPolygon'))
  65
  66
  67         def ST_DWithin(self, other: SaColumn, distance: SaColumn) -> SaColumn:
  68             return sa.func.ST_DWithin(self, other, distance, type_=sa.Float)
  69
  70
  71         def ST_Distance(self, other: SaColumn) -> SaColumn:
  72             return sa.func.ST_Distance(self, other, type_=sa.Float)
  73
  74
  75         def ST_Contains(self, other: SaColumn) -> SaColumn:
  76             return sa.func.ST_Contains(self, other, type_=sa.Float)
  77
  78
  79         def ST_ClosestPoint(self, other: SaColumn) -> SaColumn:
  80             return sa.func.ST_ClosestPoint(self, other, type_=Geometry)
  81
  82
  83         def ST_Buffer(self, other: SaColumn) -> SaColumn:
  84             return sa.func.ST_Buffer(self, other, type_=Geometry)
  85
  86
  87         def ST_Expand(self, other: SaColumn) -> SaColumn:
  88             return sa.func.ST_Expand(self, other, type_=Geometry)
  89
  90
  91         def ST_Collect(self) -> SaColumn:
  92             return sa.func.ST_Collect(self, type_=Geometry)
  93
  94
  95         def ST_Centroid(self) -> SaColumn:
  96             return sa.func.ST_Centroid(self, type_=Geometry)
  97
  98
  99         def ST_LineInterpolatePoint(self, other: SaColumn) -> SaColumn:
 100             return sa.func.ST_LineInterpolatePoint(self, other, type_=Geometry)
 101
 102
 103         def ST_LineLocatePoint(self, other: SaColumn) -> SaColumn:
 104             return sa.func.ST_LineLocatePoint(self, other, type_=sa.Float)