Re: Handling notices, warnings, etc.

"Joel Nothman" <[email protected]>
Newsgroups gmane.comp.python.db.psycopg.devel
Message-ID <[email protected]> 
I have tried to implement a callback mechanism as per below through a new  
variable connection.notice_handler.

Note that I have never dealt with the C API for Python before, so I might  
have done it all wrong:

At first I had troubles with Python threads (none was available to make  
the call to connection.notice_handler). If I have understood the nature of  
the threading problem correctly, I solved these by replacing  
Py_BEGIN_ALLOW_THREADS with gstate = PyGILState_Ensure(), etc.

I have not tested the implementation thorougly, or in a complex  
multithreaded environment.

I don't know what the common way is to pass around git data without having  
my own public repository, so I've attached a diff from revision:  
1242720889 398b3781a59ff3c36f48eb2855e9243842e94ac8

Please comment. I think this feature makes notices a lot more useful, and  
it allows for feedback from running scripts.

Thanks,

- Joel



On Sat, 30 May 2009 18:59:16 +1000, Joel Nothman  
<[email protected]> wrote:

> The suggestion is to use connection.notices. I would like to see notices  
> appear as they are output *during* execution of a query. Do the notices  
> only become available after the query execution is completed, or would  
> it be possible to check the connection.notices properly in a parallel  
> thread during execution?
>
> Thanks again,
>
> - Joel

_______________________________________________
Psycopg mailing list
Psycopg-IAPFreCvJWPBWskQ1e/[email protected]
http://lists.initd.org/mailman/listinfo/psycopg
notice_handler.diff (application/octet-stream, 10.8 KB) 
diff --git a/psycopg/connection.h b/psycopg/connection.h
index 701f46e..a0886ba 100644
--- a/psycopg/connection.h
+++ b/psycopg/connection.h
@@ -72,6 +72,7 @@ typedef struct {
     /* notice processing */
     PyObject *notice_list;
     PyObject *notice_filter;
+    PyObject *notice_handler; /* Optional callback */
     struct connectionObject_notice *notice_pending;
 
     /* notifies */
diff --git a/psycopg/connection_int.c b/psycopg/connection_int.c
index 8b54bad..8465d62 100644
--- a/psycopg/connection_int.c
+++ b/psycopg/connection_int.c
@@ -58,6 +58,10 @@ conn_notice_callback(void *args, const char *message)
         notice->message = strdup(message);
         notice->next = self->notice_pending;
         self->notice_pending = notice;
+
+        if (self->notice_handler != Py_None) {
+            PyObject_CallFunction(self->notice_handler, "s", message);
+        }
     }
 }
 
diff --git a/psycopg/connection_type.c b/psycopg/connection_type.c
index 4bba61c..cc31594 100644
--- a/psycopg/connection_type.c
+++ b/psycopg/connection_type.c
@@ -406,6 +406,7 @@ static struct PyMemberDef connectionObject_members[] = {
         "The current client encoding."},
     {"notices", T_OBJECT, offsetof(connectionObject, notice_list), RO},
     {"notifies", T_OBJECT, offsetof(connectionObject, notifies), RO},
+    {"notice_handler", T_OBJECT, offsetof(connectionObject, notice_handler), 0},
     {"dsn", T_STRING, offsetof(connectionObject, dsn), RO,
         "The current connection string."},
     {"status", T_INT,
@@ -469,6 +470,7 @@ connection_setup(connectionObject *self, const char *dsn)
     self->string_types = PyDict_New();
     self->binary_types = PyDict_New();
     self->notice_pending = NULL;
+    self->notice_handler = Py_None;
 
     pthread_mutex_init(&(self->lock), NULL);
 
@@ -512,6 +514,7 @@ connection_dealloc(PyObject* obj)
     Py_CLEAR(self->async_cursor);
     Py_CLEAR(self->notice_list);
     Py_CLEAR(self->notice_filter);
+    Py_CLEAR(self->notice_handler);
     Py_CLEAR(self->notifies);
     Py_CLEAR(self->string_types);
     Py_CLEAR(self->binary_types);
@@ -563,6 +566,7 @@ connection_traverse(connectionObject *self, visitproc visit, void *arg)
     Py_VISIT(self->async_cursor);
     Py_VISIT(self->notice_list);
     Py_VISIT(self->notice_filter);
+    Py_VISIT(self->notice_handler);
     Py_VISIT(self->notifies);
     Py_VISIT(self->string_types);
     Py_VISIT(self->binary_types);
diff --git a/psycopg/pqpath.c b/psycopg/pqpath.c
index c687ef8..e850cb7 100644
--- a/psycopg/pqpath.c
+++ b/psycopg/pqpath.c
@@ -418,7 +418,8 @@ pq_commit(connectionObject *conn)
         return 0;
     }
 
-    Py_BEGIN_ALLOW_THREADS;
+    PyGILState_STATE gstate;
+    gstate = PyGILState_Ensure();
     pthread_mutex_lock(&conn->lock);
     conn->mark += 1;
 
@@ -426,7 +427,7 @@ pq_commit(connectionObject *conn)
     retvalue = pq_execute_command_locked(conn, "COMMIT", &pgres, &error);
 
     pthread_mutex_unlock(&conn->lock);
-    Py_END_ALLOW_THREADS;
+    PyGILState_Release(gstate);;
     
     conn_notice_process(conn);
 
@@ -482,13 +483,14 @@ pq_abort(connectionObject *conn)
         return 0;
     }
 
-    Py_BEGIN_ALLOW_THREADS;
+    PyGILState_STATE gstate;
+    gstate = PyGILState_Ensure();
     pthread_mutex_lock(&conn->lock);
 
     retvalue = pq_abort_locked(conn, &pgres, &error);
 
     pthread_mutex_unlock(&conn->lock);
-    Py_END_ALLOW_THREADS;
+    PyGILState_Release(gstate);;
     
     conn_notice_process(conn);
 
@@ -515,13 +517,14 @@ pq_is_busy(connectionObject *conn)
 
     Dprintf("pq_is_busy: consuming input");
 
-    Py_BEGIN_ALLOW_THREADS;
+    PyGILState_STATE gstate;
+    gstate = PyGILState_Ensure();
     pthread_mutex_lock(&(conn->lock));
 
     if (PQconsumeInput(conn->pgconn) == 0) {
         Dprintf("pq_is_busy: PQconsumeInput() failed");
         pthread_mutex_unlock(&(conn->lock));
-        Py_BLOCK_THREADS;
+        PyGILState_Release(gstate);
         PyErr_SetString(OperationalError, PQerrorMessage(conn->pgconn));
         return -1;
     }
@@ -534,19 +537,19 @@ pq_is_busy(connectionObject *conn)
         Dprintf("curs_is_busy: got NOTIFY from pid %d, msg = %s",
                 (int) pgn->be_pid, pgn->relname);
 
-        Py_BLOCK_THREADS;
+        PyGILState_Release(gstate);
         notify = PyTuple_New(2);
         PyTuple_SET_ITEM(notify, 0, PyInt_FromLong((long)pgn->be_pid));
         PyTuple_SET_ITEM(notify, 1, PyString_FromString(pgn->relname));
         PyList_Append(conn->notifies, notify);
-        Py_UNBLOCK_THREADS;
+        gstate = PyGILState_Ensure();;
         free(pgn);
     }
 
     res = PQisBusy(conn->pgconn);
     
     pthread_mutex_unlock(&(conn->lock));
-    Py_END_ALLOW_THREADS;
+    PyGILState_Release(gstate);;
     
     conn_notice_process(conn);
 
@@ -579,12 +582,13 @@ pq_execute(cursorObject *curs, const char *query, int async)
     }
     Dprintf("curs_execute: pg connection at %p OK", curs->conn->pgconn);
 
-    Py_BEGIN_ALLOW_THREADS;
+    PyGILState_STATE gstate;
+    gstate = PyGILState_Ensure();
     pthread_mutex_lock(&(curs->conn->lock));
 
     if (pq_begin_locked(curs->conn, &pgres, &error) < 0) {
         pthread_mutex_unlock(&(curs->conn->lock));
-        Py_BLOCK_THREADS;
+        PyGILState_Release(gstate);
         pq_complete_error(curs->conn, &pgres, &error);
         return -1;
     }
@@ -598,7 +602,7 @@ pq_execute(cursorObject *curs, const char *query, int async)
         /* dont let pgres = NULL go to pq_fetch() */
         if (curs->pgres == NULL) {
             pthread_mutex_unlock(&(curs->conn->lock));
-            Py_BLOCK_THREADS;
+            PyGILState_Release(gstate);
             PyErr_SetString(OperationalError,
                             PQerrorMessage(curs->conn->pgconn));
             return -1;
@@ -618,7 +622,7 @@ pq_execute(cursorObject *curs, const char *query, int async)
         IFCLEARPGRES(curs->pgres);
         if (PQsendQuery(curs->conn->pgconn, query) == 0) {
             pthread_mutex_unlock(&(curs->conn->lock));
-            Py_BLOCK_THREADS;
+            PyGILState_Release(gstate);
             PyErr_SetString(OperationalError,
                             PQerrorMessage(curs->conn->pgconn));
             return -1;
@@ -627,7 +631,7 @@ pq_execute(cursorObject *curs, const char *query, int async)
     }
 
     pthread_mutex_unlock(&(curs->conn->lock));
-    Py_END_ALLOW_THREADS;
+    PyGILState_Release(gstate);
 
     conn_notice_process(curs->conn);
 
@@ -663,7 +667,8 @@ _pq_fetch_tuples(cursorObject *curs)
     int pgnfields;
     int pgbintuples;
 
-    Py_BEGIN_ALLOW_THREADS;
+    PyGILState_STATE gstate;
+    gstate = PyGILState_Ensure();
     pthread_mutex_lock(&(curs->conn->lock));
 
     pgnfields = PQnfields(curs->pgres);
@@ -672,20 +677,20 @@ _pq_fetch_tuples(cursorObject *curs)
     curs->notuples = 0;
 
     /* create the tuple for description and typecasting */
-    Py_BLOCK_THREADS;
+    PyGILState_Release(gstate);;
     Py_XDECREF(curs->description);
     Py_XDECREF(curs->casts);    
     curs->description = PyTuple_New(pgnfields);
     curs->casts = PyTuple_New(pgnfields);
     curs->columns = pgnfields;
-    Py_UNBLOCK_THREADS;
+    gstate = PyGILState_Ensure();;
 
     /* calculate the display size for each column (cpu intensive, can be
        switched off at configuration time) */
 #ifdef PSYCOPG_DISPLAY_SIZE
-    Py_BLOCK_THREADS;
+    PyGILState_Release(gstate);;
     dsize = (int *)PyMem_Malloc(pgnfields * sizeof(int));
-    Py_UNBLOCK_THREADS;
+    gstate = PyGILState_Ensure();;
     if (dsize != NULL) {
         int j, len;
         for (i=0; i < pgnfields; i++) {
@@ -710,7 +715,7 @@ _pq_fetch_tuples(cursorObject *curs)
         PyObject *type;
         PyObject *cast = NULL;
 
-        Py_BLOCK_THREADS;
+        PyGILState_Release(gstate);;
 
         dtitem = PyTuple_New(7);
         PyTuple_SET_ITEM(curs->description, i, dtitem);
@@ -799,17 +804,17 @@ _pq_fetch_tuples(cursorObject *curs)
         Py_INCREF(Py_None);
         PyTuple_SET_ITEM(dtitem, 6, Py_None);
     
-        Py_UNBLOCK_THREADS;    
+        gstate = PyGILState_Ensure();;    
     }
 
     if (dsize) {
-        Py_BLOCK_THREADS;
+        PyGILState_Release(gstate);;
         PyMem_Free(dsize);
-        Py_UNBLOCK_THREADS;
+        gstate = PyGILState_Ensure();;
    }
    
     pthread_mutex_unlock(&(curs->conn->lock));
-    Py_END_ALLOW_THREADS;
+    PyGILState_Release(gstate);;
 }
 
 #ifdef HAVE_PQPROTOCOL3
@@ -831,7 +836,8 @@ _pq_copy_in_v3(cursorObject *curs)
         }
         if (length == 0 || length > INT_MAX || error == 1) break;
 
-        Py_BEGIN_ALLOW_THREADS;
+        PyGILState_STATE gstate;
+        gstate = PyGILState_Ensure();
         res = PQputCopyData(curs->conn->pgconn, PyString_AS_STRING(o),
             /* Py_ssize_t->int cast was validated above */
             (int) length);
@@ -847,7 +853,7 @@ _pq_copy_in_v3(cursorObject *curs)
                 PQerrorMessage(curs->conn->pgconn));
             error = 2;
         }
-        Py_END_ALLOW_THREADS;
+        PyGILState_Release(gstate);;
 
         if (error == 2) break;
 
@@ -936,9 +942,11 @@ _pq_copy_out_v3(cursorObject *curs)
     Py_ssize_t len;
 
     while (1) {
-        Py_BEGIN_ALLOW_THREADS;
+        PyGILState_STATE gstate;
+        gstate = PyGILState_Ensure();
+
         len = PQgetCopyData(curs->conn->pgconn, &buffer, 0);
-        Py_END_ALLOW_THREADS;
+        PyGILState_Release(gstate);;
 
         if (len > 0 && buffer) {
             tmp = PyObject_CallMethod(curs->copyfile,
@@ -981,9 +989,10 @@ _pq_copy_out(cursorObject *curs)
     Py_ssize_t len;
 
     while (1) {
-        Py_BEGIN_ALLOW_THREADS;
+        PyGILState_STATE gstate;
+        gstate = PyGILState_Ensure();
         status = PQgetline(curs->conn->pgconn, buffer, 4096);
-        Py_END_ALLOW_THREADS;
+        PyGILState_Release(gstate);;
         if (status == 0) {
             if (!ll && buffer[0] == '\\' && buffer[1] == '.') break;
 
@@ -1043,7 +1052,8 @@ pq_fetch(cursorObject *curs)
             struct timeval tv;
             int sval, sock;
 
-            Py_BEGIN_ALLOW_THREADS;
+            PyGILState_STATE gstate;
+            gstate = PyGILState_Ensure();
             pthread_mutex_lock(&(curs->conn->lock));
 
             sock = PQsocket(curs->conn->pgconn);
@@ -1061,10 +1071,11 @@ pq_fetch(cursorObject *curs)
             sval = select(sock+1, &rfds, NULL, NULL, &tv);
 
             pthread_mutex_unlock(&(curs->conn->lock));
-            Py_END_ALLOW_THREADS;
+            PyGILState_Release(gstate);;
         }
 
-        Py_BEGIN_ALLOW_THREADS;
+        PyGILState_STATE gstate;
+        gstate = PyGILState_Ensure();
         pthread_mutex_lock(&(curs->conn->lock));
 
         Dprintf("pq_fetch: data is probably ready");
@@ -1072,7 +1083,7 @@ pq_fetch(cursorObject *curs)
         curs->pgres = PQgetResult(curs->conn->pgconn);
 
         pthread_mutex_unlock(&(curs->conn->lock));
-        Py_END_ALLOW_THREADS;
+        PyGILState_Release(gstate);;
     }
 
     /* check for PGRES_FATAL_ERROR result */
lmpx.com only provides a reader for public news (NNTP) servers. It is not affiliated with the servers or forums shown here and is not responsible for the content of articles, which is written by their respective authors.