Fix query encoding

postgresql_proxy/interceptors.py

@@ -101,7 +101,8 @@ def _intercept_context_data(self, data):
     def _intercept_query(self, query, interceptors):
         logging.getLogger("intercept").debug("intercepting query\n%s", query)
         # Remove zero byte at the end
-        query = query[:-1].decode("utf-8")
+        codec = self.get_codec()
+        query = query[:-1].decode(codec)
         for interceptor in interceptors:
             func = self._get_plugin_interceptor_function(interceptor)
             query = func(query, self.context)
@@ -113,7 +114,7 @@ def _intercept_query(self, query, interceptors):
             )
 
         # Append the zero byte at the end
-        return query.encode("utf-8") + b"\x00"
+        return query.encode(codec) + b"\x00"
 
 
 class ResponseInterceptor(Interceptor):

tests/test_proxy.py

@@ -332,3 +332,26 @@ def test_psql_ssl_file_batch_stress_no_hang(postgres_settings, ssl_proxy_port):
                     "psql -f batch succeeded but expected marker missing "
                     f"(run={run_idx + 1}, {elapsed=:.2f}s) stdout_tail={out_tail}"
                 )
+
+
+def test_non_utf8_client_encoding_query_text_through_proxy(
+    postgres_settings, plain_proxy_port
+):
+    """Proxy query interception should honor client_encoding when decoding query text."""
+    with psycopg2.connect(
+        host="127.0.0.1",
+        port=plain_proxy_port,
+        user=postgres_settings["user"],
+        password=postgres_settings["password"],
+        dbname=postgres_settings["dbname"],
+        sslmode="disable",
+        connect_timeout=3,
+        client_encoding="LATIN1",
+    ) as conn:
+        conn.autocommit = True
+        with conn.cursor() as cur:
+            cur.execute("SHOW client_encoding")
+            assert cur.fetchone() == ("LATIN1",)
+
+            cur.execute("SELECT 'olá'::text")
+            assert cur.fetchone() == ("olá",)