Fixed #30148 -- Logged COPY ... TO statements in connection.queries on PostgreSQL.

django/db/backends/postgresql/base.py

@@ -11,6 +11,9 @@ from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
 from django.db import connections
 from django.db.backends.base.base import BaseDatabaseWrapper
+from django.db.backends.utils import (
+    CursorDebugWrapper as BaseCursorDebugWrapper,
+)
 from django.db.utils import DatabaseError as WrappedDatabaseError
 from django.utils.functional import cached_property
 from django.utils.safestring import SafeString
@@ -281,3 +284,16 @@ class DatabaseWrapper(BaseDatabaseWrapper):
     def pg_version(self):
         with self.temporary_connection():
             return self.connection.server_version
+
+    def make_debug_cursor(self, cursor):
+        return CursorDebugWrapper(cursor, self)
+
+
+class CursorDebugWrapper(BaseCursorDebugWrapper):
+    def copy_expert(self, sql, file, *args):
+        with self.debug_sql(sql):
+            return self.cursor.copy_expert(sql, file, *args)
+
+    def copy_to(self, file, table, *args, **kwargs):
+        with self.debug_sql(sql='COPY %s TO STDOUT' % table):
+            return self.cursor.copy_to(file, table, *args, **kwargs)

docs/releases/3.0.txt

@@ -201,6 +201,8 @@ Models
   :class:`~django.db.models.functions.Trunc` database functions determines the
   treatment of nonexistent and ambiguous datetimes.
 
+* ``connection.queries`` now shows ``COPY … TO`` statements on PostgreSQL.
+
 Requests and Responses
 ~~~~~~~~~~~~~~~~~~~~~~
 

tests/backends/postgresql/tests.py

 import unittest
+from io import StringIO
 from unittest import mock
 
 from django.core.exceptions import ImproperlyConfigured
 from django.db import DatabaseError, connection, connections
-from django.test import TestCase
+from django.test import TestCase, override_settings
 
 
 @unittest.skipUnless(connection.vendor == 'postgresql', 'PostgreSQL tests')
@@ -176,3 +177,15 @@ class Tests(TestCase):
             self.assertEqual(psycopg2_version(), (4, 2, 1))
         with mock.patch('psycopg2.__version__', '4.2b0.dev1 (dt dec pq3 ext lo64)'):
             self.assertEqual(psycopg2_version(), (4, 2))
+
+    @override_settings(DEBUG=True)
+    def test_copy_cursors(self):
+        out = StringIO()
+        copy_expert_sql = 'COPY django_session TO STDOUT (FORMAT CSV, HEADER)'
+        with connection.cursor() as cursor:
+            cursor.copy_expert(copy_expert_sql, out)
+            cursor.copy_to(out, 'django_session')
+        self.assertEqual(
+            [q['sql'] for q in connection.queries],
+            [copy_expert_sql, 'COPY django_session TO STDOUT'],
+        )