Fix #337 Bug when passing a query to execute that is a psycopg2 SQL object

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'):

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'):