diff --git a/cacheops/transaction.py b/cacheops/transaction.py index 7c6ffe64..bb3045ed 100644 --- a/cacheops/transaction.py +++ b/cacheops/transaction.py @@ -114,6 +114,9 @@ def is_sql_dirty(sql): # but some people will pass it anyway if isinstance(sql, bytes): sql = sql.decode() + # Handle psycopg2/psycopg3 sql.Composed/sql.SQL objects (see #377) + elif not isinstance(sql, str): + sql = str(sql) # NOTE: not using regex here for speed sql = sql.lower() for action in ('update', 'insert', 'delete'): diff --git a/tests/tests_transactions.py b/tests/tests_transactions.py index 5d01a8d4..055e6e62 100644 --- a/tests/tests_transactions.py +++ b/tests/tests_transactions.py @@ -1,12 +1,19 @@ +import unittest + from django.db import connection, IntegrityError from django.db.transaction import atomic from django.test import TransactionTestCase -from cacheops.transaction import queue_when_in_transaction +from cacheops.transaction import is_sql_dirty, queue_when_in_transaction from .models import Category, Post from .utils import run_in_thread +try: + from psycopg2 import sql +except ImportError: + sql = None + def get_category(): return Category.objects.cache().get(pk=1) @@ -121,6 +128,24 @@ def cacheops_commit_handler(using): self.assertEqual(calls, ['cacheops', 'django']) + @unittest.skipUnless(sql, "psycopg2 not installed") + def test_is_sql_dirty_with_composed_objects(self): + """sql.Composed/sql.SQL objects should not crash is_sql_dirty (#377).""" + composed = sql.SQL("DELETE FROM {}").format(sql.Identifier("some_table")) + simple = sql.SQL("SELECT * FROM foo") + + # These are not strings — calling .lower() on them would raise AttributeError + self.assertNotIsInstance(composed, str) + self.assertNotIsInstance(simple, str) + self.assertFalse(hasattr(composed, "lower")) + self.assertFalse(hasattr(simple, "lower")) + + # is_sql_dirty should handle them without raising + self.assertTrue(is_sql_dirty(composed)) + self.assertTrue(is_sql_dirty(sql.SQL("INSERT INTO foo VALUES (%s)"))) + self.assertTrue(is_sql_dirty(sql.SQL("UPDATE foo SET bar = %s"))) + self.assertFalse(is_sql_dirty(simple)) + def test_multidb(self): try: with atomic('slave'):