1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210
|
from django.contrib.gis.db.models import GeometryField
from django.contrib.gis.db.models.functions import Distance
from django.contrib.gis.measure import Area as AreaMeasure
from django.contrib.gis.measure import Distance as DistanceMeasure
from django.db import NotSupportedError
from django.utils.functional import cached_property
class BaseSpatialOperations:
# Quick booleans for the type of this spatial backend, and
# an attribute for the spatial database version tuple (if applicable)
postgis = False
spatialite = False
mariadb = False
mysql = False
oracle = False
spatial_version = None
# How the geometry column should be selected.
select = "%s"
@cached_property
def select_extent(self):
return self.select
# Aggregates
disallowed_aggregates = ()
geom_func_prefix = ""
# Mapping between Django function names and backend names, when names do not
# match; used in spatial_function_name().
function_names = {}
# Set of known unsupported functions of the backend
unsupported_functions = {
"Area",
"AsGeoJSON",
"AsGML",
"AsKML",
"AsSVG",
"Azimuth",
"BoundingCircle",
"Centroid",
"ClosestPoint",
"Difference",
"Distance",
"Envelope",
"FromWKB",
"FromWKT",
"GeoHash",
"GeometryDistance",
"Intersection",
"IsEmpty",
"IsValid",
"Length",
"LineLocatePoint",
"MakeValid",
"MemSize",
"NumGeometries",
"NumPoints",
"Perimeter",
"PointOnSurface",
"Reverse",
"Scale",
"SnapToGrid",
"SymDifference",
"Transform",
"Translate",
"Union",
}
# Constructors
from_text = False
# Default conversion functions for aggregates; will be overridden if implemented
# for the spatial backend.
def convert_extent(self, box, srid):
raise NotImplementedError(
"Aggregate extent not implemented for this spatial backend."
)
def convert_extent3d(self, box, srid):
raise NotImplementedError(
"Aggregate 3D extent not implemented for this spatial backend."
)
# For quoting column values, rather than columns.
def geo_quote_name(self, name):
return "'%s'" % name
# GeometryField operations
def geo_db_type(self, f):
"""
Return the database column type for the geometry field on
the spatial backend.
"""
raise NotImplementedError(
"subclasses of BaseSpatialOperations must provide a geo_db_type() method"
)
def get_distance(self, f, value, lookup_type):
"""
Return the distance parameters for the given geometry field,
lookup value, and lookup type.
"""
raise NotImplementedError(
"Distance operations not available on this spatial backend."
)
def get_geom_placeholder(self, f, value, compiler):
"""
Return the placeholder for the given geometry field with the given
value. Depending on the spatial backend, the placeholder may contain a
stored procedure call to the transformation function of the spatial
backend.
"""
def transform_value(value, field):
return value is not None and value.srid != field.srid
if hasattr(value, "as_sql"):
return (
"%s(%%s, %s)" % (self.spatial_function_name("Transform"), f.srid)
if transform_value(value.output_field, f)
else "%s"
)
if transform_value(value, f):
# Add Transform() to the SQL placeholder.
return "%s(%s(%%s,%s), %s)" % (
self.spatial_function_name("Transform"),
self.from_text,
value.srid,
f.srid,
)
elif self.connection.features.has_spatialrefsys_table:
return "%s(%%s,%s)" % (self.from_text, f.srid)
else:
# For backwards compatibility on MySQL (#27464).
return "%s(%%s)" % self.from_text
def check_expression_support(self, expression):
if isinstance(expression, self.disallowed_aggregates):
raise NotSupportedError(
"%s spatial aggregation is not supported by this database backend."
% expression.name
)
super().check_expression_support(expression)
def spatial_aggregate_name(self, agg_name):
raise NotImplementedError(
"Aggregate support not implemented for this spatial backend."
)
def spatial_function_name(self, func_name):
if func_name in self.unsupported_functions:
raise NotSupportedError(
"This backend doesn't support the %s function." % func_name
)
return self.function_names.get(func_name, self.geom_func_prefix + func_name)
# Routines for getting the OGC-compliant models.
def geometry_columns(self):
raise NotImplementedError(
"Subclasses of BaseSpatialOperations must provide a geometry_columns() "
"method."
)
def spatial_ref_sys(self):
raise NotImplementedError(
"subclasses of BaseSpatialOperations must a provide spatial_ref_sys() "
"method"
)
distance_expr_for_lookup = staticmethod(Distance)
def get_db_converters(self, expression):
converters = super().get_db_converters(expression)
if isinstance(expression.output_field, GeometryField):
converters.append(self.get_geometry_converter(expression))
return converters
def get_geometry_converter(self, expression):
raise NotImplementedError(
"Subclasses of BaseSpatialOperations must provide a "
"get_geometry_converter() method."
)
def get_area_att_for_field(self, field):
if field.geodetic(self.connection):
if self.connection.features.supports_area_geodetic:
return "sq_m"
raise NotImplementedError(
"Area on geodetic coordinate systems not supported."
)
else:
units_name = field.units_name(self.connection)
if units_name:
return AreaMeasure.unit_attname(units_name)
def get_distance_att_for_field(self, field):
dist_att = None
if field.geodetic(self.connection):
if self.connection.features.supports_distance_geodetic:
dist_att = "m"
else:
units = field.units_name(self.connection)
if units:
dist_att = DistanceMeasure.unit_attname(units)
return dist_att
|