mirror of
https://github.com/django/django.git
synced 2025-10-07 05:49:30 +00:00
217 lines
6.8 KiB
Python
217 lines
6.8 KiB
Python
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",
|
|
"AsWKB",
|
|
"AsWKT",
|
|
"Azimuth",
|
|
"BoundingCircle",
|
|
"Centroid",
|
|
"ClosestPoint",
|
|
"Difference",
|
|
"Distance",
|
|
"DistanceSpheroid",
|
|
"Envelope",
|
|
"ForcePolygonCW",
|
|
"FromWKB",
|
|
"FromWKT",
|
|
"GeoHash",
|
|
"GeometryDistance",
|
|
"GeometryType",
|
|
"Intersection",
|
|
"IsEmpty",
|
|
"IsValid",
|
|
"Length",
|
|
"LineLocatePoint",
|
|
"MakeValid",
|
|
"MemSize",
|
|
"NumGeometries",
|
|
"NumPoints",
|
|
"Perimeter",
|
|
"PointOnSurface",
|
|
"Reverse",
|
|
"Rotate",
|
|
"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
|