[4.2.x] Fixed #33638 -- Fixed GIS lookups crash with geography fields on PostGIS.

Backport of 4403432b75 from main
This commit is contained in:
Jacob Walls 2023-02-02 10:23:16 -05:00 committed by Mariusz Felisiak
parent 600b88db4c
commit 714d59d57f
3 changed files with 59 additions and 27 deletions

View File

@ -27,7 +27,8 @@ BILATERAL = "bilateral"
class PostGISOperator(SpatialOperator): class PostGISOperator(SpatialOperator):
def __init__(self, geography=False, raster=False, **kwargs): def __init__(self, geography=False, raster=False, **kwargs):
# Only a subset of the operators and functions are available for the # Only a subset of the operators and functions are available for the
# geography type. # geography type. Lookups that don't support geography will be cast to
# geometry.
self.geography = geography self.geography = geography
# Only a subset of the operators and functions are available for the # Only a subset of the operators and functions are available for the
# raster type. Lookups that don't support raster will be converted to # raster type. Lookups that don't support raster will be converted to
@ -37,13 +38,8 @@ class PostGISOperator(SpatialOperator):
super().__init__(**kwargs) super().__init__(**kwargs)
def as_sql(self, connection, lookup, template_params, *args): def as_sql(self, connection, lookup, template_params, *args):
if lookup.lhs.output_field.geography and not self.geography:
raise ValueError(
'PostGIS geography does not support the "%s" '
"function/operator." % (self.func or self.op,)
)
template_params = self.check_raster(lookup, template_params) template_params = self.check_raster(lookup, template_params)
template_params = self.check_geography(lookup, template_params)
return super().as_sql(connection, lookup, template_params, *args) return super().as_sql(connection, lookup, template_params, *args)
def check_raster(self, lookup, template_params): def check_raster(self, lookup, template_params):
@ -93,6 +89,12 @@ class PostGISOperator(SpatialOperator):
return template_params return template_params
def check_geography(self, lookup, template_params):
"""Convert geography fields to geometry types, if necessary."""
if lookup.lhs.output_field.geography and not self.geography:
template_params["lhs"] += "::geometry"
return template_params
class ST_Polygon(Func): class ST_Polygon(Func):
function = "ST_Polygon" function = "ST_Polygon"

View File

@ -18,6 +18,16 @@ class City(NamedModel):
app_label = "geogapp" app_label = "geogapp"
class CityUnique(NamedModel):
point = models.PointField(geography=True, unique=True)
class Meta:
required_db_features = {
"supports_geography",
"supports_geometry_field_unique_index",
}
class Zipcode(NamedModel): class Zipcode(NamedModel):
code = models.CharField(max_length=10) code = models.CharField(max_length=10)
poly = models.PolygonField(geography=True) poly = models.PolygonField(geography=True)

View File

@ -6,12 +6,14 @@ import os
from django.contrib.gis.db import models from django.contrib.gis.db import models
from django.contrib.gis.db.models.functions import Area, Distance from django.contrib.gis.db.models.functions import Area, Distance
from django.contrib.gis.measure import D from django.contrib.gis.measure import D
from django.core.exceptions import ValidationError
from django.db import NotSupportedError, connection from django.db import NotSupportedError, connection
from django.db.models.functions import Cast from django.db.models.functions import Cast
from django.test import TestCase, skipIfDBFeature, skipUnlessDBFeature from django.test import TestCase, skipIfDBFeature, skipUnlessDBFeature
from django.test.utils import CaptureQueriesContext
from ..utils import FuncTestMixin from ..utils import FuncTestMixin
from .models import City, County, Zipcode from .models import City, CityUnique, County, Zipcode
class GeographyTest(TestCase): class GeographyTest(TestCase):
@ -38,28 +40,46 @@ class GeographyTest(TestCase):
for cities in [cities1, cities2]: for cities in [cities1, cities2]:
self.assertEqual(["Dallas", "Houston", "Oklahoma City"], cities) self.assertEqual(["Dallas", "Houston", "Oklahoma City"], cities)
def test04_invalid_operators_functions(self): @skipUnlessDBFeature("supports_geography", "supports_geometry_field_unique_index")
def test_geography_unique(self):
""" """
Exceptions are raised for operators & functions invalid on geography Cast geography fields to geometry type when validating uniqueness to
fields. remove the reliance on unavailable ~= operator.
""" """
if not connection.ops.postgis:
self.skipTest("This is a PostGIS-specific test.")
# Only a subset of the geometry functions & operator are available
# to PostGIS geography types. For more information, visit:
# http://postgis.refractions.net/documentation/manual-1.5/ch08.html#PostGIS_GeographyFunctions
z = Zipcode.objects.get(code="77002")
# ST_Within not available.
with self.assertRaises(ValueError):
City.objects.filter(point__within=z.poly).count()
# `@` operator not available.
with self.assertRaises(ValueError):
City.objects.filter(point__contained=z.poly).count()
# Regression test for #14060, `~=` was never really implemented for PostGIS.
htown = City.objects.get(name="Houston") htown = City.objects.get(name="Houston")
with self.assertRaises(ValueError): CityUnique.objects.create(point=htown.point)
City.objects.get(point__exact=htown.point) duplicate = CityUnique(point=htown.point)
msg = "City unique with this Point already exists."
with self.assertRaisesMessage(ValidationError, msg):
duplicate.validate_unique()
@skipUnlessDBFeature("supports_geography")
def test_operators_functions_unavailable_for_geography(self):
"""
Geography fields are cast to geometry if the relevant operators or
functions are not available.
"""
z = Zipcode.objects.get(code="77002")
point_field = "%s.%s::geometry" % (
connection.ops.quote_name(City._meta.db_table),
connection.ops.quote_name("point"),
)
# ST_Within.
qs = City.objects.filter(point__within=z.poly)
with CaptureQueriesContext(connection) as ctx:
self.assertEqual(qs.count(), 1)
self.assertIn(f"ST_Within({point_field}", ctx.captured_queries[0]["sql"])
# @ operator.
qs = City.objects.filter(point__contained=z.poly)
with CaptureQueriesContext(connection) as ctx:
self.assertEqual(qs.count(), 1)
self.assertIn(f"{point_field} @", ctx.captured_queries[0]["sql"])
# ~= operator.
htown = City.objects.get(name="Houston")
qs = City.objects.filter(point__exact=htown.point)
with CaptureQueriesContext(connection) as ctx:
self.assertEqual(qs.count(), 1)
self.assertIn(f"{point_field} ~=", ctx.captured_queries[0]["sql"])
def test05_geography_layermapping(self): def test05_geography_layermapping(self):
"Testing LayerMapping support on models with geography fields." "Testing LayerMapping support on models with geography fields."