django/django

View on GitHub
django/contrib/gis/db/backends/base/operations.py

Summary

Maintainability
A
1 hr
Test Coverage
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, Distance as DistanceMeasure,
)
from django.db.utils 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
    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

    # Does the spatial database have a geometry or geography type?
    geography = False
    geometry = False

    # 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 = {}

    # Blacklist/set of known unsupported functions of the backend
    unsupported_functions = {
        'Area', 'AsGeoJSON', 'AsGML', 'AsKML', 'AsSVG', 'Azimuth',
        'BoundingCircle', 'Centroid', 'Difference', 'Distance', 'Envelope',
        'GeoHash', 'Intersection', '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