From b3bd9738f2ba4aa50c074c059882f1b1cf19035e Mon Sep 17 00:00:00 2001
From: Daniele Varrazzo <daniele.varrazzo@gmail.com>
Date: Tue, 17 Mar 2020 23:46:08 +1300
Subject: [PATCH] Raise a few errors in pg implementation module

Better to make this interface a bit thicker instead of forcing user to
write Python as it was C.
---
 psycopg3/pq/__init__.py  |  2 ++
 psycopg3/pq/pq_ctypes.py | 25 ++++++++++++++++++++-----
 tests/pq/test_async.py   |  7 +++----
 3 files changed, 25 insertions(+), 9 deletions(-)

diff --git a/psycopg3/pq/__init__.py b/psycopg3/pq/__init__.py
index 641610137..3bbe1587e 100644
--- a/psycopg3/pq/__init__.py
+++ b/psycopg3/pq/__init__.py
@@ -17,6 +17,7 @@ from .enums import (
     Ping,
     DiagnosticField,
 )
+from .encodings import py_codecs
 
 from . import pq_ctypes as pq_module
 
@@ -34,4 +35,5 @@ __all__ = (
     "PGconn",
     "Conninfo",
     "PQerror",
+    "py_codecs",
 )
diff --git a/psycopg3/pq/pq_ctypes.py b/psycopg3/pq/pq_ctypes.py
index 7502f25b8..a916e5547 100644
--- a/psycopg3/pq/pq_ctypes.py
+++ b/psycopg3/pq/pq_ctypes.py
@@ -153,6 +153,14 @@ class PGconn:
     def error_message(self):
         return impl.PQerrorMessage(self.pgconn_ptr)
 
+    @property
+    def error_str(self):
+        rv = self.error_message
+        if rv:
+            return rv.encode('utf8', 'replace').rstrip()
+        else:
+            return "no details available"
+
     @property
     def socket(self):
         return impl.PQsocket(self.pgconn_ptr)
@@ -337,19 +345,26 @@ class PGconn:
         return PGresult(rv) if rv else None
 
     def consume_input(self):
-        return impl.PQconsumeInput(self.pgconn_ptr)
+        if 1 != impl.PQconsumeInput(self.pgconn_ptr):
+            raise PQerror(f"consuming input failed: {self.error_str}")
 
     def is_busy(self):
         return impl.PQisBusy(self.pgconn_ptr)
 
-    def is_non_blocking(self):
+    @property
+    def nonblocking(self):
         return impl.PQisnonblocking(self.pgconn_ptr)
 
-    def set_non_blocking(self, arg):
-        return impl.PQsetnonblocking(self.pgconn_ptr, arg)
+    @nonblocking.setter
+    def nonblocking(self, arg):
+        if 0 > impl.PQsetnonblocking(self.pgconn_ptr, arg):
+            raise PQerror(f"setting nonblocking failed: {self.error_str}")
 
     def flush(self):
-        return impl.PQflush(self.pgconn_ptr)
+        rv = impl.PQflush(self.pgconn_ptr)
+        if rv < 0:
+            raise PQerror(f"flushing failed: {self.error_str}")
+        return rv
 
 
 class PGresult:
diff --git a/tests/pq/test_async.py b/tests/pq/test_async.py
index be2971c98..730011dc2 100644
--- a/tests/pq/test_async.py
+++ b/tests/pq/test_async.py
@@ -3,7 +3,7 @@ from select import select
 
 def test_send_query(pq, pgconn):
     # This test shows how to process an async query in all its glory
-    pgconn.set_non_blocking(1)
+    pgconn.nonblocking = 1
 
     # Long query to make sure we have to wait on send
     pgconn.send_query(
@@ -15,7 +15,6 @@ def test_send_query(pq, pgconn):
     waited_on_send = 0
     while 1:
         f = pgconn.flush()
-        assert f != -1
         if f == 0:
             break
 
@@ -26,7 +25,7 @@ def test_send_query(pq, pgconn):
         if wl:
             continue  # call flush again()
         if rl:
-            assert pgconn.consume_input() == 1, pgconn.error_message
+            pgconn.consume_input()
             continue
 
     assert waited_on_send
@@ -34,7 +33,7 @@ def test_send_query(pq, pgconn):
     # read loop
     results = []
     while 1:
-        assert pgconn.consume_input() == 1, pgconn.error_message
+        pgconn.consume_input()
         if pgconn.is_busy():
             select([pgconn.socket], [], [])
             continue
-- 
2.47.3