""" # noqa: E501
+ render_bind_cast = True
+
+ def adapt(self, impltype):
+ """dynamically adapt a range type to an abstract impl.
+
+ For example ``INT4RANGE().adapt(_Psycopg2NumericRange)`` should
+ produce a type that will have ``_Psycopg2NumericRange`` behaviors
+ and also render as ``INT4RANGE`` in SQL and DDL.
+
+ """
+ if issubclass(impltype, AbstractRangeImpl):
+ # two ways to do this are: 1. create a new type on the fly
+ # or 2. have AbstractRangeImpl(visit_name) constructor and a
+ # visit_abstract_range_impl() method in the PG compiler.
+ # I'm choosing #1 as the resulting type object
+ # will then make use of the same mechanics
+ # as if we had made all these sub-types explicitly, and will
+ # also look more obvious under pdb etc.
+ # The adapt() operation here is cached per type-class-per-dialect,
+ # so is not much of a performance concern
+ visit_name = self.__visit_name__
+ return type(
+ f"{visit_name}RangeImpl",
+ (impltype, self.__class__),
+ {"__visit_name__": visit_name},
+ )()
+ else:
+ return super().adapt(impltype)
+
class comparator_factory(sqltypes.Concatenable.Comparator):
"""Define comparison operations for range types."""
return self.expr.op("+")(other)
+class AbstractRangeImpl(AbstractRange):
+ """marker for AbstractRange that will apply a subclass-specific
+ adaptation"""
+
+
class AbstractMultiRange(AbstractRange):
"""base for PostgreSQL MULTIRANGE types"""
+class AbstractMultiRangeImpl(AbstractRangeImpl, AbstractMultiRange):
+ """marker for AbstractRange that will apply a subclass-specific
+ adaptation"""
+
+
class INT4RANGE(AbstractRange):
"""Represent the PostgreSQL INT4RANGE type."""
sqltypes.BOOLEANTYPE,
)
+ def test_where_equal_obj(self):
+ self._test_clause(
+ self.col == self._data_obj(),
+ f"data_table.range = %(range_1)s::{self._col_str}",
+ sqltypes.BOOLEANTYPE,
+ )
+
def test_where_not_equal(self):
self._test_clause(
self.col != self._data_str(),
sqltypes.BOOLEANTYPE,
)
+ def test_where_not_equal_obj(self):
+ self._test_clause(
+ self.col != self._data_obj(),
+ f"data_table.range <> %(range_1)s::{self._col_str}",
+ sqltypes.BOOLEANTYPE,
+ )
+
def test_where_is_null(self):
self._test_clause(
self.col == None, "data_table.range IS NULL", sqltypes.BOOLEANTYPE
sqltypes.BOOLEANTYPE,
)
+ def test_contains_obj(self):
+ self._test_clause(
+ self.col.contains(self._data_obj()),
+ f"data_table.range @> %(range_1)s::{self._col_str}",
+ sqltypes.BOOLEANTYPE,
+ )
+
def test_contained_by(self):
self._test_clause(
self.col.contained_by(self._data_str()),
)
cls.col = table.c.range
+ def test_auto_cast_back_to_type(self, connection):
+ """test that a straight pass of the range type without any context
+ will send appropriate casting info so that the driver can round
+ trip it.
+
+ This doesn't happen in general across other backends and not for
+ types like JSON etc., although perhaps it should, as we now have
+ pretty straightforward infrastructure to turn it on; asyncpg
+ for example does cast JSONs now in place. But that's a
+ bigger issue; for PG ranges it's likely useful to do this for
+ PG backends as this is a fairly narrow use case.
+
+ Brought up in #8540.
+
+ """
+ data_obj = self._data_obj()
+ stmt = select(literal(data_obj, type_=self._col_type))
+ round_trip = connection.scalar(stmt)
+ eq_(round_trip, data_obj)
+
def test_actual_type(self):
eq_(str(self._col_type()), self._col_str)
sqltypes.BOOLEANTYPE,
)
+ def test_where_equal_obj(self):
+ self._test_clause(
+ self.col == self._data_obj(),
+ f"data_table.multirange = %(multirange_1)s::{self._col_str}",
+ sqltypes.BOOLEANTYPE,
+ )
+
def test_where_not_equal(self):
self._test_clause(
self.col != self._data_str(),
sqltypes.BOOLEANTYPE,
)
+ def test_where_not_equal_obj(self):
+ self._test_clause(
+ self.col != self._data_obj(),
+ f"data_table.multirange <> %(multirange_1)s::{self._col_str}",
+ sqltypes.BOOLEANTYPE,
+ )
+
def test_where_is_null(self):
self._test_clause(
self.col == None,
sqltypes.BOOLEANTYPE,
)
+ def test_contained_by_obj(self):
+ self._test_clause(
+ self.col.contained_by(self._data_obj()),
+ f"data_table.multirange <@ %(multirange_1)s::{self._col_str}",
+ sqltypes.BOOLEANTYPE,
+ )
+
def test_overlaps(self):
self._test_clause(
self.col.overlaps(self._data_str()),
sqltypes.BOOLEANTYPE,
)
+ def test_adjacent_to_obj(self):
+ self._test_clause(
+ self.col.adjacent_to(self._data_obj()),
+ f"data_table.multirange -|- %(multirange_1)s::{self._col_str}",
+ sqltypes.BOOLEANTYPE,
+ )
+
def test_union(self):
self._test_clause(
self.col + self.col,
)
cls.col = table.c.range
+ def test_auto_cast_back_to_type(self, connection):
+ """test that a straight pass of the range type without any context
+ will send appropriate casting info so that the driver can round
+ trip it.
+
+ This doesn't happen in general across other backends and not for
+ types like JSON etc., although perhaps it should, as we now have
+ pretty straightforward infrastructure to turn it on; asyncpg
+ for example does cast JSONs now in place. But that's a
+ bigger issue; for PG ranges it's likely useful to do this for
+ PG backends as this is a fairly narrow use case.
+
+ Brought up in #8540.
+
+ """
+ data_obj = self._data_obj()
+ stmt = select(literal(data_obj, type_=self._col_type))
+ round_trip = connection.scalar(stmt)
+ eq_(round_trip, data_obj)
+
def test_actual_type(self):
eq_(str(self._col_type()), self._col_str)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
