--- /dev/null
+.. change::
+ :tags: bug, sql, regression
+ :tickets: 6436
+
+ The :class:`.TypeDecorator` class will now emit a warning when used in SQL
+ compilation with caching unless the ``.cache_ok`` flag is set to ``True``
+ or ``False``. A new class-level attribute :attr:`.TypeDecorator.cache_ok`
+ may be set which will be used as an indication that all the parameters
+ passed to the object are safe to be used as a cache key if set to ``True``,
+ ``False`` means they are not.
class TZDateTime(TypeDecorator):
impl = DateTime
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value is not None:
"""
impl = CHAR
+ cache_ok = True
def load_dialect_impl(self, dialect):
if dialect.name == 'postgresql':
impl = VARCHAR
+ cache_ok = True
+
def process_bind_param(self, value, dialect):
if value is not None:
value = json.dumps(value)
impl = VARCHAR
+ cache_ok = True
+
def coerce_compared_value(self, op, value):
if op in (operators.like_op, operators.not_like_op):
return String()
class PGPString(TypeDecorator):
impl = BYTEA
+ cache_ok = True
+
def __init__(self, passphrase):
super(PGPString, self).__init__()
+
self.passphrase = passphrase
def bind_expression(self, bindvalue):
class CoerceUnicode(TypeDecorator):
impl = Unicode
+ cache_ok = True
def process_bind_param(self, value, dialect):
if util.py2k and isinstance(value, util.binary_type):
correct value as string.
"""
impl = Unicode
+ cache_ok = True
def column_expression(self, colexpr):
return cast(colexpr, Numeric)
class MixedBinary(TypeDecorator):
impl = String
+ cache_ok = True
def process_result_value(self, value, dialect):
if isinstance(value, str):
"""
impl = LargeBinary
+ cache_ok = True
def __init__(
self, protocol=pickle.HIGHEST_PROTOCOL, pickler=None, comparator=None
impl = DateTime
epoch = dt.datetime.utcfromtimestamp(0)
+ cache_ok = True
def __init__(self, native=True, second_precision=None, day_precision=None):
"""Construct an Interval object.
# efficient switch construct
if meth is STATIC_CACHE_KEY:
- result += (attrname, obj._static_cache_key)
+ sck = obj._static_cache_key
+ if sck is NO_CACHE:
+ anon_map[NO_CACHE] = True
+ return None
+ result += (attrname, sck)
elif meth is ANON_NAME:
elements = util.preloaded.sql_elements
if isinstance(obj, elements._anonymous_label):
"""
+from sqlalchemy.sql.traversals import NO_CACHE
from . import operators
from .base import SchemaEventTarget
from .visitors import Traversible
impl = types.Unicode
+ cache_ok = True
+
def process_bind_param(self, value, dialect):
return "PREFIX:" + value
given; in this case, the ``impl`` variable can reference
``TypeEngine`` as a placeholder.
+ The :attr:`.TypeDecorator.cache_ok` class-level flag indicates if this
+ custom :class:`.TypeDecorator` is safe to be used as part of a cache key.
+ This flag defaults to ``None`` which will initially generate a warning
+ when the SQL compiler attempts to generate a cache key for a statement
+ that uses this type. If the :class:`.TypeDecorator` is not guaranteed
+ to produce the same bind/result behavior and SQL generation
+ every time, this flag should be set to ``False``; otherwise if the
+ class produces the same behavior each time, it may be set to ``True``.
+ See :attr:`.TypeDecorator.cache_ok` for further notes on how this works.
+
Types that receive a Python type that isn't similar to the ultimate type
used may want to define the :meth:`TypeDecorator.coerce_compared_value`
method. This is used to give the expression system a hint when coercing
class MyJsonType(TypeDecorator):
impl = postgresql.JSON
+ cache_ok = True
+
def coerce_compared_value(self, op, value):
return self.impl.coerce_compared_value(op, value)
"""
+ cache_ok = None
+ """Indicate if statements using this :class:`.TypeDecorator` are "safe to
+ cache".
+
+ The default value ``None`` will emit a warning and then not allow caching
+ of a statement which includes this type. Set to ``False`` to disable
+ statements using this type from being cached at all without a warning.
+ When set to ``True``, the object's class and selected elements from its
+ state will be used as part of the cache key, e.g.::
+
+ class MyType(TypeDecorator):
+ impl = String
+
+ cache_ok = True
+
+ def __init__(self, choices):
+ self.choices = tuple(choices)
+ self.internal_only = True
+
+ The cache key for the above type would be equivalent to::
+
+ (<class '__main__.MyType'>, ('choices', ('a', 'b', 'c')))
+
+ The caching scheme will extract attributes from the type that correspond
+ to the names of parameters in the ``__init__()`` method. Above, the
+ "choices" attribute becomes part of the cache key but "internal_only"
+ does not, because there is no parameter named "internal_only".
+
+ The requirements for cacheable elements is that they are hashable
+ and also that they indicate the same SQL rendered for expressions using
+ this type every time for a given cache value.
+
+ .. versionadded:: 1.4.14 - added the ``cache_ok`` flag to allow
+ some configurability of caching for :class:`.TypeDecorator` classes.
+
+ .. seealso::
+
+ :ref:`sql_caching`
+
+ """
+
class Comparator(TypeEngine.Comparator):
"""A :class:`.TypeEngine.Comparator` that is specific to
:class:`.TypeDecorator`.
{},
)
+ @property
+ def _static_cache_key(self):
+ if self.cache_ok is None:
+ util.warn(
+ "TypeDecorator %r will not produce a cache key because "
+ "the ``cache_ok`` flag is not set to True. "
+ "Set this flag to True if this type object's "
+ "state is safe to use in a cache key, or False to "
+ "disable this warning." % self
+ )
+ elif self.cache_ok is True:
+ return super(TypeDecorator, self)._static_cache_key
+
+ return NO_CACHE
+
def _gen_dialect_impl(self, dialect):
"""
#todo
"""
+ cache_ok = True
+
def __init__(self, base, mapping):
"""Construct a new :class:`.Variant`.
def define_tables(cls, metadata):
class Decorated(TypeDecorator):
impl = cls.datatype
+ cache_ok = True
Table(
"date_table",
def string_as_int(self):
class StringAsInt(TypeDecorator):
impl = String(50)
+ cache_ok = True
def get_dbapi_type(self, dbapi):
return dbapi.NUMBER
class MyPickleType(types.TypeDecorator):
impl = PickleType
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value:
def test_cast_type_decorator(self):
class MyInteger(sqltypes.TypeDecorator):
impl = Integer
+ cache_ok = True
type_ = MyInteger()
t = sql.table("t", sql.column("col"))
class MyTime(TypeDecorator):
impl = TIMESTAMP
+ cache_ok = True
@testing.combinations(
(TIMESTAMP,),
def test_limit_preserves_typing_information(self):
class MyType(TypeDecorator):
impl = Integer
+ cache_ok = True
stmt = select(type_coerce(column("x"), MyType).label("foo")).limit(1)
dialect = oracle.dialect()
class TestTypeDec(TypeDecorator):
impl = NullType()
+ cache_ok = True
def load_dialect_impl(self, dialect):
if dialect.name == "oracle":
class BITD(TypeDecorator):
impl = Integer
+ cache_ok = True
+
def load_dialect_impl(self, dialect):
if dialect.name == "postgresql":
return BigInteger()
class SpecialType(sqltypes.TypeDecorator):
impl = String
+ cache_ok = True
+
def process_bind_param(self, value, dialect):
return value + " processed"
def test_custom_subclass(self, metadata, connection):
class MyEnum(TypeDecorator):
impl = Enum("oneHI", "twoHI", "threeHI", name="myenum")
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value is not None:
def define_tables(cls, metadata):
class ProcValue(TypeDecorator):
impl = cls.ARRAY(Integer, dimensions=2)
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value is None:
class _ArrayOfEnum(TypeDecorator):
# previous workaround for array of enum
impl = postgresql.ARRAY
+ cache_ok = True
def bind_expression(self, bindvalue):
return sa.cast(bindvalue, self)
def test_sqlite_autoincrement_int_affinity(self):
class MyInteger(sqltypes.TypeDecorator):
impl = Integer
+ cache_ok = True
table = Table(
"autoinctable",
class SpecialType(sqltypes.TypeDecorator):
impl = String
+ cache_ok = True
def process_bind_param(self, value, dialect):
return value + " processed"
def test_exception_wrapping_non_dbapi_statement(self):
class MyType(TypeDecorator):
impl = Integer
+ cache_ok = True
def process_bind_param(self, value, dialect):
raise SomeException("nope")
class MyType(TypeDecorator):
impl = Integer
+ cache_ok = True
def process_bind_param(self, value, dialect):
raise MyException("nope")
class MyType(TypeDecorator):
impl = Integer
+ cache_ok = True
def process_bind_param(self, value, dialect):
raise nope
class IntToStr(TypeDecorator[int]):
impl = String
+ cache_ok = True
def process_bind_param(
self,
class JSONEncodedDict(TypeDecorator):
impl = VARCHAR(50)
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value is not None:
class JSONEncodedDict(TypeDecorator):
impl = VARCHAR(50)
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value is not None:
class JSONEncodedDict(TypeDecorator):
impl = VARCHAR(50)
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value is not None:
class JSONEncodedDict(TypeDecorator):
impl = VARCHAR(50)
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value is not None:
class IntDecorator(TypeDecorator):
impl = Integer
+ cache_ok = True
class SmallintDecorator(TypeDecorator):
impl = SmallInteger
+ cache_ok = True
class SomeDBInteger(sa.Integer):
pass
def _unhashable_fixture(self, metadata, load_on_pending=False):
class MyHashType(sa.TypeDecorator):
impl = sa.String(100)
+ cache_ok = True
def process_bind_param(self, value, dialect):
return ";".join(
class StringAsInt(TypeDecorator):
impl = String(50)
+ cache_ok = True
def get_dbapi_type(self, dbapi):
return dbapi.NUMBER
def define_tables(cls, metadata):
class MyUnsortable(TypeDecorator):
impl = String(10)
+ cache_ok = True
def process_bind_param(self, value, dialect):
return "%s,%s" % (value["x"], value["y"])
class MyType(TypeDecorator):
impl = Integer
hashable = False
+ cache_ok = True
def process_result_value(self, value, dialect):
return [value]
def define_tables(cls, metadata):
class MySpecialType(sa.types.TypeDecorator):
impl = String
+ cache_ok = True
def process_bind_param(self, value, dialect):
return "lala" + value
class MyType(TypeDecorator):
impl = String(50)
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value is not None:
class EvalsNull(TypeDecorator):
impl = String(50)
+ cache_ok = True
+ cache_ok = True
should_evaluate_none = True
def define_tables(cls, metadata):
class SpecialType(TypeDecorator):
impl = Date
+ cache_ok = True
def process_bind_param(self, value, dialect):
assert isinstance(value, datetime.date)
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy import tuple_
+from sqlalchemy import TypeDecorator
from sqlalchemy import union
from sqlalchemy import union_all
from sqlalchemy import util
from sqlalchemy.testing import is_not
from sqlalchemy.testing import is_true
from sqlalchemy.testing import ne_
+from sqlalchemy.testing.assertions import expect_warnings
from sqlalchemy.testing.util import random_choices
from sqlalchemy.types import ARRAY
from sqlalchemy.types import JSON
dml.Delete.argument_for("sqlite", "foo", None)
+class MyType1(TypeDecorator):
+ cache_ok = True
+ impl = String
+
+
+class MyType2(TypeDecorator):
+ cache_ok = True
+ impl = Integer
+
+
+class MyType3(TypeDecorator):
+ impl = Integer
+
+ cache_ok = True
+
+ def __init__(self, arg):
+ self.arg = arg
+
+
class CoreFixtures(object):
# lambdas which return a tuple of ColumnElement objects.
# must return at least two objects that should compare differently.
lambda: (table_a, table_b),
]
+ type_cache_key_fixtures = [
+ lambda: (
+ column("q") == column("x"),
+ column("q") == column("y"),
+ column("z") == column("x"),
+ column("z", String(50)) == column("x", String(50)),
+ column("z", String(50)) == column("x", String(30)),
+ column("z", String(50)) == column("x", Integer),
+ column("z", MyType1()) == column("x", MyType2()),
+ column("z", MyType1()) == column("x", MyType3("x")),
+ column("z", MyType1()) == column("x", MyType3("y")),
+ )
+ ]
+
dont_compare_values_fixtures = [
lambda: (
# note the in_(...) all have different column names because
for fixtures_, compare_values in [
(self.fixtures, True),
(self.dont_compare_values_fixtures, False),
+ (self.type_cache_key_fixtures, False),
]:
for fixture in fixtures_:
self._run_cache_key_fixture(fixture, compare_values)
is_true(case.is_select)
else:
is_false(case.is_select)
+
+
+class TypesTest(fixtures.TestBase):
+ def test_typedec_no_cache(self):
+ class MyType(TypeDecorator):
+ impl = String
+
+ expr = column("q", MyType()) == 1
+
+ with expect_warnings(
+ r"TypeDecorator MyType\(\) will not produce a cache key"
+ ):
+ is_(expr._generate_cache_key(), None)
+
+ def test_typedec_cache_false(self):
+ class MyType(TypeDecorator):
+ impl = String
+
+ cache_ok = False
+
+ expr = column("q", MyType()) == 1
+
+ is_(expr._generate_cache_key(), None)
+
+ def test_typedec_cache_ok(self):
+ class MyType(TypeDecorator):
+ impl = String
+
+ cache_ok = True
+
+ def go1():
+ expr = column("q", MyType()) == 1
+ return expr
+
+ def go2():
+ expr = column("p", MyType()) == 1
+ return expr
+
+ c1 = go1()._generate_cache_key()[0]
+ c2 = go1()._generate_cache_key()[0]
+ c3 = go2()._generate_cache_key()[0]
+
+ eq_(c1, c2)
+ ne_(c1, c3)
+
+ def test_typedec_cache_ok_params(self):
+ class MyType(TypeDecorator):
+ impl = String
+
+ cache_ok = True
+
+ def __init__(self, p1, p2):
+ self.p1 = p1
+ self._p2 = p2
+
+ def go1():
+ expr = column("q", MyType("x", "y")) == 1
+ return expr
+
+ def go2():
+ expr = column("q", MyType("q", "y")) == 1
+ return expr
+
+ def go3():
+ expr = column("q", MyType("x", "z")) == 1
+ return expr
+
+ c1 = go1()._generate_cache_key()[0]
+ c2 = go1()._generate_cache_key()[0]
+ c3 = go2()._generate_cache_key()[0]
+ c4 = go3()._generate_cache_key()[0]
+
+ eq_(c1, c2)
+ ne_(c1, c3)
+ eq_(c1, c4)
class MyType(TypeDecorator):
impl = String(50)
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value is not None:
def test_autoinc_detection_no_affinity(self):
class MyType(TypeDecorator):
impl = TypeEngine
+ cache_ok = True
assert MyType()._type_affinity is None
t = Table("x", MetaData(), Column("id", MyType(), primary_key=True))
class MyInteger(TypeDecorator):
impl = Integer
+ cache_ok = True
+
def process_bind_param(self, value, dialect):
if value is None:
return None
class MyTypeDecAndSchema(TypeDecorator, sqltypes.SchemaType):
impl = String()
+ cache_ok = True
evt_targets = ()
class MyType(TypeDecorator):
impl = target_typ
+ cache_ok = True
typ = MyType()
self._test_before_parent_attach(typ, target_typ)
def test_before_parent_attach_typedec_of_schematype(self):
class MyType(TypeDecorator, sqltypes.SchemaType):
impl = String
+ cache_ok = True
typ = MyType()
self._test_before_parent_attach(typ)
def test_before_parent_attach_schematype_of_typedec(self):
class MyType(sqltypes.SchemaType, TypeDecorator):
impl = String
+ cache_ok = True
typ = MyType()
self._test_before_parent_attach(typ)
def test_to_metadata_copy_decorated(self):
class MyDecorated(TypeDecorator):
impl = self.MyType
+ cache_ok = True
m1 = MetaData()
def _add_override_factory(self):
class MyInteger(TypeDecorator):
impl = Integer
+ cache_ok = True
class comparator_factory(TypeDecorator.Comparator):
def __init__(self, expr):
def _add_override_factory(self):
class MyIntegerOne(TypeDecorator):
impl = Integer
+ cache_ok = True
class comparator_factory(TypeDecorator.Comparator):
def __init__(self, expr):
class MyIntegerTwo(TypeDecorator):
impl = MyIntegerOne
+ cache_ok = True
return MyIntegerTwo
class MyInteger(TypeDecorator):
impl = Integer
+ cache_ok = True
class comparator_factory(TypeDecorator.Comparator):
def __init__(self, expr):
class MyDecInteger(TypeDecorator):
impl = MyInteger
+ cache_ok = True
return MyDecInteger
class MyInteger(TypeDecorator):
impl = Integer
+ cache_ok = True
def process_bind_param(self, value, dialect):
return int(value[4:])
class NameWithProcess(TypeDecorator):
impl = String
+ cache_ok = True
def process_bind_param(self, value, dialect):
return value[3:]
class Goofy1(TypeDecorator):
impl = String
+ cache_ok = True
def process_result_value(self, value, dialect):
return value + "a"
class Goofy2(TypeDecorator):
impl = String
+ cache_ok = True
def process_result_value(self, value, dialect):
return value + "b"
class Goofy3(TypeDecorator):
impl = String
+ cache_ok = True
def process_result_value(self, value, dialect):
return value + "c"
def _test_result_processor(self, cls, use_cache):
class MyType(TypeDecorator):
impl = String()
+ cache_ok = True
def process_result_value(self, value, dialect):
return "HI " + value
def define_tables(cls, metadata):
class GoofyType(TypeDecorator):
impl = String
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value is None:
def test_select_doesnt_pollute_result(self, connection):
class MyType(TypeDecorator):
impl = Integer
+ cache_ok = True
def process_result_value(self, value, dialect):
raise Exception("I have not been selected")
def test_type_coerce_preserve_subq(self):
class MyType(TypeDecorator):
impl = Integer
+ cache_ok = True
stmt = select(type_coerce(column("x"), MyType).label("foo"))
subq = stmt.subquery()
def _type_decorator_outside_fixture(self):
class MyString(TypeDecorator):
impl = String
+ cache_ok = True
def bind_expression(self, bindvalue):
return func.outside_bind(bindvalue)
class MyString(TypeDecorator):
impl = MyInsideString
+ cache_ok = True
return self._test_table(MyString)
class MyString(TypeDecorator):
impl = String
+ cache_ok = True
# this works because when the compiler calls dialect_impl(),
# a copy of MyString is created which has just this impl
def define_tables(cls, metadata):
class MyString(TypeDecorator):
impl = String
+ cache_ok = True
def bind_expression(self, bindvalue):
return func.lower(bindvalue)
class MyType(TypeDecorator):
impl = CHAR
+ cache_ok = True
def load_dialect_impl(self, dialect):
if dialect.name == "postgresql":
class MyDecoratedType(types.TypeDecorator):
impl = String
+ cache_ok = True
def bind_processor(self, dialect):
impl_processor = super(MyDecoratedType, self).bind_processor(
class MyNewUnicodeType(types.TypeDecorator):
impl = Unicode
+ cache_ok = True
def process_bind_param(self, value, dialect):
return "BIND_IN" + value
class MyNewIntType(types.TypeDecorator):
impl = Integer
+ cache_ok = True
def process_bind_param(self, value, dialect):
return value * 10
class MyUnicodeType(types.TypeDecorator):
impl = Unicode
+ cache_ok = True
def bind_processor(self, dialect):
impl_processor = super(MyUnicodeType, self).bind_processor(
class MyDecOfDec(types.TypeDecorator):
impl = MyNewIntType
+ cache_ok = True
Table(
"users",
def test_typedecorator_literal_render(self):
class MyType(types.TypeDecorator):
impl = String
+ cache_ok = True
def process_literal_param(self, value, dialect):
return "HI->%s<-THERE" % value
# value rendering.
class MyType(types.TypeDecorator):
impl = String
+ cache_ok = True
def process_bind_param(self, value, dialect):
return "HI->%s<-THERE" % value
class MyType(types.TypeDecorator):
impl = impl_
+ cache_ok = True
dec_type = MyType(**kw)
def test_user_defined_typedec_impl(self):
class MyType(types.TypeDecorator):
impl = Float
+ cache_ok = True
def load_dialect_impl(self, dialect):
if dialect.name == "sqlite":
def test_typedecorator_schematype_constraint(self, typ):
class B(TypeDecorator):
impl = typ
+ cache_ok = True
t1 = Table("t1", MetaData(), Column("q", B(create_constraint=True)))
eq_(
class MyType(TypeDecorator):
impl = VARCHAR
+ cache_ok = True
+
eq_(repr(MyType(45)), "MyType(length=45)")
def test_user_defined_typedec_impl_bind(self):
class MyType(types.TypeDecorator):
impl = TypeOne
+ cache_ok = True
def load_dialect_impl(self, dialect):
if dialect.name == "sqlite":
def define_tables(cls, metadata):
class MyType(types.TypeDecorator):
impl = String(50)
+ cache_ok = True
def process_bind_param(self, value, dialect):
return "BIND_IN" + str(value)
def test_type_decorator_variant_one_roundtrip(self, variant_roundtrip):
class Foo(TypeDecorator):
impl = String(50)
+ cache_ok = True
if testing.against("postgresql"):
data = [5, 6, 10]
class Foo(TypeDecorator):
impl = variant
+ cache_ok = True
if testing.against("postgresql"):
data = assert_data = [5, 6, 10]
def test_type_decorator_variant_three(self, variant_roundtrip):
class Foo(TypeDecorator):
impl = String
+ cache_ok = True
if testing.against("postgresql"):
data = ["five", "six", "ten"]
def test_type_decorator_compile_variant_one(self):
class Foo(TypeDecorator):
impl = String
+ cache_ok = True
self.assert_compile(
Foo().with_variant(Integer, "sqlite"),
class Foo(TypeDecorator):
impl = variant
+ cache_ok = True
self.assert_compile(
Foo().with_variant(Integer, "sqlite"),
def test_type_decorator_compile_variant_three(self):
class Foo(TypeDecorator):
impl = String
+ cache_ok = True
self.assert_compile(
Integer().with_variant(Foo(), "postgresql"),
self.name = name
class MyEnum(TypeDecorator):
+ cache_ok = True
+
def __init__(self, values):
self.impl = Enum(
*[v.name for v in values],
class MyPickleType(types.TypeDecorator):
impl = PickleType
+ cache_ok = True
def process_bind_param(self, value, dialect):
if value:
class MyTypeDec(types.TypeDecorator):
impl = String
+ cache_ok = True
+
def process_bind_param(self, value, dialect):
return "BIND_IN" + str(value)
class MyDecOfDec(types.TypeDecorator):
impl = MyTypeDec
+ cache_ok = True
+
Table(
"test",
metadata,
class CoerceNothing(TypeDecorator):
coerce_to_is_types = ()
impl = Integer
+ cache_ok = True
class CoerceBool(TypeDecorator):
coerce_to_is_types = (bool,)
impl = Boolean
+ cache_ok = True
class CoerceNone(TypeDecorator):
coerce_to_is_types = (type(None),)
impl = Integer
+ cache_ok = True
c1 = column("x", CoerceNothing())
c2 = column("x", CoerceBool())
def test_typedec_righthand_coercion(self, connection):
class MyTypeDec(types.TypeDecorator):
impl = String
+ cache_ok = True
def process_bind_param(self, value, dialect):
return "BIND_IN" + str(value)
class MyBool(TypeDecorator):
impl = Boolean(create_constraint=True)
+ cache_ok = True
# future method
def process_literal_param(self, value, dialect):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
