operations.py

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