/* pqpath.c - single path into libpq
 *
 * Copyright (C) 2003 Federico Di Gregorio <fog@debian.org>
 *
 * This file is part of psycopg.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2,
 * or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

/* IMPORTANT NOTE: no function in this file do its own connection locking
   except for pg_execute and pq_fetch (that are somehow high-level. This means
   that all the othe functions should be called while holding a lock to the
   connection.
*/

#include <Python.h>
#include <string.h>

#define PSYCOPG_MODULE
#include "psycopg/config.h"
#include "psycopg/python.h"
#include "psycopg/psycopg.h"
#include "psycopg/pqpath.h"
#include "psycopg/connection.h"
#include "psycopg/cursor.h"
#include "psycopg/typecast.h"
#include "psycopg/pgtypes.h"
#include "psycopg/pgversion.h"

/* pq_raise - raise a python exception of the right kind

   This function should be called while holding the GIL. */

void
pq_raise(connectionObject *conn, cursorObject *curs, PyObject *exc, char *msg)
{
    PyObject *pgc = (PyObject*)curs;
    
    char *err = NULL;
    char *err2 = NULL;
    char *code = NULL;
    char *buf = NULL;
    
    if ((conn == NULL && curs == NULL) || (curs != NULL && conn == NULL)) {
        PyErr_SetString(Error, "psycopg went psycotic and raised a null error");
        return;
    }
    
    if (curs && curs->pgres) {
        err = PQresultErrorMessage(curs->pgres);
#ifdef HAVE_PQPROTOCOL3
        if (err != NULL && conn->protocol == 3) {
            code = PQresultErrorField(curs->pgres, PG_DIAG_SQLSTATE);
        }
#endif
    }
    if (err == NULL)
        err = PQerrorMessage(conn->pgconn);

    /* if the is no error message we probably called pq_raise without reason:
       we need to set an exception anyway because the caller will probably
       raise and a meaningful message is better than an empty one */
    if (err == NULL) {
        PyErr_SetString(Error, "psycopg went psycotic without error set");
        return;
    }
    
    /* if exc is NULL, analyze the message and try to deduce the right
       exception kind (only if we have a pgres, obviously) */
    if (exc == NULL) {
        if (curs && curs->pgres) {
            if (conn->protocol == 3) {
#ifdef HAVE_PQPROTOCOL3
                char *pgstate =
                    PQresultErrorField(curs->pgres, PG_DIAG_SQLSTATE);
                if (pgstate != NULL && !strncmp(pgstate, "23", 2))
                    exc = IntegrityError;
                else
                    exc = ProgrammingError;
#endif
            }
        }
    }
    
    /* if exc is still NULL psycopg was not built with HAVE_PQPROTOCOL3 or the
       connection is using protocol 2: in both cases we default to comparing
       error messages */
    if (exc == NULL) {
        if (!strncmp(err, "ERROR:  Cannot insert a duplicate key", 37)
            || !strncmp(err, "ERROR:  ExecAppend: Fail to add null", 36)
            || strstr(err, "referential integrity violation"))
            exc = IntegrityError;
        else
            exc = ProgrammingError;
    }
    
    /* try to remove the initial "ERROR: " part from the postgresql error */
    if (err && strlen(err) > 8) err2 = &(err[8]);
    else err2 = err;

    /* if msg is not NULL, add it to the error message, after a '\n' */
    if (msg && code) {
        int len = strlen(code) + strlen(err) + strlen(msg) + 5;
        if ((buf = PyMem_Malloc(len))) {
            snprintf(buf, len, "[%s] %s\n%s", code, err2, msg);
            psyco_set_error(exc, pgc, buf, err, code);
        }
    }
    else if (msg) {
        int len = strlen(err) + strlen(msg) + 2;
        if ((buf = PyMem_Malloc(len))) {
            snprintf(buf, len, "%s\n%s", err2, msg);
            psyco_set_error(exc, pgc, buf, err, code);
        }
    }
    else {
        psyco_set_error(exc, pgc, err2, err, code);        
    }
    
    if (buf != NULL) PyMem_Free(buf);
}

/* pq_set_critical, pq_resolve_critical - manage critical errors

   this function is invoked when a PQexec() call returns NULL, meaning a
   critical condition like out of memory or lost connection. it save the error
   message and mark the connection as 'wanting cleanup'.

   both functions do not call any Py_*_ALLOW_THREADS macros.
   pq_resolve_critical should be called while holding the GIL. */

void
pq_set_critical(connectionObject *conn, const char *msg)
{
    if (msg == NULL) 
        msg = PQerrorMessage(conn->pgconn);
    if (conn->critical) free(conn->critical);
    if (msg && msg[0] != '\0') conn->critical = strdup(msg);
    else conn->critical = NULL;
}

PyObject *
pq_resolve_critical(connectionObject *conn, int close)
{
    Dprintf("pq_resolve_critical: resolving %s", conn->critical);
    
    if (conn->critical) {
        char *msg = &(conn->critical[6]);
        Dprintf("pq_resolve_critical: error = %s", msg);
        /* we can't use pq_raise because the error has already been cleared
           from the connection, so we just raise an OperationalError with the
           critical message */
        PyErr_SetString(OperationalError, msg);
        
        /* we don't want to destroy this connection but just close it */
        if (close == 1) conn_close(conn);
    }
    return NULL;
}

/* pq_clear_async - clear the effects of a previous async query

   note that this function does block because it needs to wait for the full
   result sets of the previous query to clear them.

   
   this function does not call any Py_*_ALLOW_THREADS macros */

void
pq_clear_async(connectionObject *conn)
{
    PGresult *pgres;

    do {
        pgres = PQgetResult(conn->pgconn);
        Dprintf("pq_clear_async: clearing PGresult at %p", pgres);
        IFCLEARPGRES(pgres);
    } while (pgres != NULL);
}

/* pq_begin - send a BEGIN WORK, if necessary

   this function does not call any Py_*_ALLOW_THREADS macros */

int
pq_begin(connectionObject *conn)
{
    const char *query[] = {
        NULL,
        "BEGIN; SET TRANSACTION ISOLATION LEVEL READ COMMITTED",
        "BEGIN; SET TRANSACTION ISOLATION LEVEL SERIALIZABLE"};
    
    int pgstatus, retvalue = -1;
    PGresult *pgres = NULL;

    Dprintf("pq_begin: pgconn = %p, isolevel = %ld, status = %d",
            conn->pgconn, conn->isolation_level, conn->status);

    if (conn->isolation_level == 0 || conn->status != CONN_STATUS_READY) {
        Dprintf("pq_begin: transaction in progress");
        return 0;
    }

    pq_clear_async(conn);
    pgres = PQexec(conn->pgconn, query[conn->isolation_level]);
    if (pgres == NULL) {
        Dprintf("pq_begin: PQexec() failed");
        pq_set_critical(conn, NULL);
        goto cleanup;
    }

    pgstatus = PQresultStatus(pgres);
    if (pgstatus != PGRES_COMMAND_OK ) {
        Dprintf("pq_begin: result is NOT OK");
        pq_set_critical(conn, NULL);
        goto cleanup;
    }
    Dprintf("pq_begin: issued '%s' command", query[conn->isolation_level]);

    retvalue = 0;
    conn->status = CONN_STATUS_BEGIN;

 cleanup:
    IFCLEARPGRES(pgres);
    return retvalue;
}

/* pq_commit - send an END, if necessary

   this function does not call any Py_*_ALLOW_THREADS macros */

int
pq_commit(connectionObject *conn)
{
    const char *query = "END";
    int pgstatus, retvalue = -1;
    PGresult *pgres = NULL;

    Dprintf("pq_commit: pgconn = %p, isolevel = %ld, status = %d",
            conn->pgconn, conn->isolation_level, conn->status);

    if (conn->isolation_level == 0 || conn->status != CONN_STATUS_BEGIN) {
        Dprintf("pq_commit: no transaction to commit");
        return 0;
    }

    pq_clear_async(conn);
    pgres = PQexec(conn->pgconn, query);
    if (pgres == NULL) {
        Dprintf("pq_commit: PQexec() failed");
        pq_set_critical(conn, NULL);
        goto cleanup;
    }

    pgstatus = PQresultStatus(pgres);
    if (pgstatus != PGRES_COMMAND_OK ) {
        Dprintf("pq_commit: result is NOT OK");
        pq_set_critical(conn, NULL);
        goto cleanup;
    }
    Dprintf("pq_commit: issued '%s' command", query);

    retvalue = 0;
    conn->status = CONN_STATUS_READY;

 cleanup:
    IFCLEARPGRES(pgres);
    return retvalue;
}

/* pq_abort - send an ABORT, if necessary

   this function does not call any Py_*_ALLOW_THREADS macros */

int
pq_abort(connectionObject *conn)
{
    const char *query = "ABORT";
    int pgstatus, retvalue = -1;
    PGresult *pgres = NULL;

    Dprintf("pq_abort: pgconn = %p, isolevel = %ld, status = %d",
            conn->pgconn, conn->isolation_level, conn->status);

    if (conn->isolation_level == 0 || conn->status != CONN_STATUS_BEGIN) {
        Dprintf("pq_abort: no transaction to abort");
        return 0;
    }

    pq_clear_async(conn);
    pgres = PQexec(conn->pgconn, query);
    if (pgres == NULL) {
        Dprintf("pq_abort: PQexec() failed");
        pq_set_critical(conn, NULL);
        goto cleanup;
    }

    pgstatus = PQresultStatus(pgres);
    if (pgstatus != PGRES_COMMAND_OK ) {
        Dprintf("pq_abort: result is NOT OK");
        pq_set_critical(conn, NULL);
        goto cleanup;
    }
    Dprintf("pq_abort: issued '%s' command", query);

    retvalue = 0;
    conn->status = CONN_STATUS_READY;

 cleanup:
    IFCLEARPGRES(pgres);
    return retvalue;
}

/* pq_is_busy - consume input and return connection status
 
   a status of 1 means that a call to pq_fetch will block, while a status of 0
   means that there is data available to be collected. -1 means an error, the
   exception will be set accordingly.

   this fucntion locks the connection object
   this function call Py_*_ALLOW_THREADS macros */

int
pq_is_busy(connectionObject *conn)
{
    PGnotify *pgn;
    
    Dprintf("pq_is_busy: consuming input");

    Py_BEGIN_ALLOW_THREADS;
    pthread_mutex_lock(&(conn->lock));

    if (PQconsumeInput(conn->pgconn) == 0) {
        Dprintf("pq_is_busy: PQconsumeInput() failed");
        pthread_mutex_unlock(&(conn->lock));
        Py_BLOCK_THREADS;
        PyErr_SetString(OperationalError, PQerrorMessage(conn->pgconn));
        return -1;
    }

    pthread_mutex_unlock(&(conn->lock));
    Py_END_ALLOW_THREADS;
    
    /* now check for notifies */
    while ((pgn = PQnotifies(conn->pgconn)) != NULL) {
        PyObject *notify;
        
        Dprintf("curs_is_busy: got NOTIFY from pid %d, msg = %s",
                pgn->be_pid, pgn->relname);

        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);
        free(pgn);
    }
    
    return PQisBusy(conn->pgconn);
}

/* pq_execute - execute a query, possibly asyncronously

   this fucntion locks the connection object
   this function call Py_*_ALLOW_THREADS macros */

int
pq_execute(cursorObject *curs, const char *query, int async)
{
    /* if the status of the connection is critical raise an exception and
       definitely close the connection */
    if (curs->conn->critical) {
        pq_resolve_critical(curs->conn, 1);
        return -1;
    }

    /* check status of connection, raise error if not OK */
    if (PQstatus(curs->conn->pgconn) != CONNECTION_OK) {
        Dprintf("pq_execute: connection NOT OK");
        PyErr_SetString(OperationalError, PQerrorMessage(curs->conn->pgconn));
        return -1;
    }
    Dprintf("curs_execute: pg connection at %p OK", curs->conn->pgconn);

    Py_BEGIN_ALLOW_THREADS;
    pthread_mutex_lock(&(curs->conn->lock));

    pq_begin(curs->conn);

    if (async == 0) {
        IFCLEARPGRES(curs->pgres);
        Dprintf("pq_execute: executing SYNC query:");
        Dprintf("    %-.200s", query);
        curs->pgres = PQexec(curs->conn->pgconn, query);
    }

    else if (async == 1) {
        /* first of all, let see if the previous query has already ended, if
           not what should we do? just block and discard data or execute
           another query? */
        pq_clear_async(curs->conn);
        
        Dprintf("pq_execute: executing ASYNC query:");
        Dprintf("    %-.200s", query);
        
        /* then we can go on and send a new query without fear */
        IFCLEARPGRES(curs->pgres);
        if (PQsendQuery(curs->conn->pgconn, query) == 0) {
            pthread_mutex_unlock(&(curs->conn->lock));
            Py_BLOCK_THREADS;
            PyErr_SetString(OperationalError,
                            PQerrorMessage(curs->conn->pgconn));
            return -1;
        }
        Dprintf("pq_execute: async query sent to backend");
    }
    
    pthread_mutex_unlock(&(curs->conn->lock));
    Py_END_ALLOW_THREADS;
    
    /* if the execute was sync, we call pq_fetch() immediately,
       to respect the old DBAPI-2.0 compatible behaviour */
    if (async == 0) {
        Dprintf("pq_execute: entering syncronous DBAPI compatibility mode");
        if (pq_fetch(curs) == -1) return -1;
    }
    else {
        curs->conn->async_cursor = (PyObject*)curs;
    }
    
    return 1-async;
}


/* pq_fetch - fetch data after a query

   this fucntion locks the connection object
   this function call Py_*_ALLOW_THREADS macros

   return value:
     -1 - some error occurred while calling libpq
      0 - no result from the backend but no libpq errors
      1 - result from backend (possibly data is ready)
*/

static void
_pq_fetch_tuples(cursorObject *curs)
{
    int i, *dsize = NULL;

    int pgnfields = PQnfields(curs->pgres);
    int pgbintuples = PQbinaryTuples(curs->pgres);

    curs->notuples = 0;

    /* create the tuple for description and typecasting */
    Py_XDECREF(curs->description);
    Py_XDECREF(curs->casts);
    curs->description = PyTuple_New(pgnfields);
    curs->casts = PyTuple_New(pgnfields);
    curs->columns = pgnfields;
    
    /* calculate the display size for each column (cpu intensive, can be
       switched off at configuration time) */
#ifdef PSYCOPG_DISPLAY_SIZE
    dsize = (int *)PyMem_Malloc(pgnfields * sizeof(int));
    if (dsize != NULL) {
        int j, len;
        for (i=0; i < pgnfields; i++) {
            dsize[i] = -1;
        }
        for (j = 0; j < curs->rowcount; j++) {
            for (i = 0; i < pgnfields; i++) {
                len = PQgetlength(curs->pgres, j, i);
                if (len > dsize[i]) dsize[i] = len;
            }
        }
    }
#endif

    /* calculate various parameters and typecasters */
    for (i = 0; i < pgnfields; i++) {
        Oid ftype = PQftype(curs->pgres, i);
        int fsize = PQfsize(curs->pgres, i);
        int fmod =  PQfmod(curs->pgres, i);
        
        PyObject *dtitem = PyTuple_New(7);
        PyObject *type = PyInt_FromLong(ftype);
        PyObject *cast = NULL;
        
        PyTuple_SET_ITEM(curs->description, i, dtitem);
        
        /* fill the right cast function by accessing the global dictionary of
           casting objects.  if we got no defined cast use the default
           one */
        if (!(cast = PyDict_GetItem(curs->casts, type))) {
            Dprintf("_pq_fetch_tuples: cast %d not in per-cursor dict", ftype);
            if (!(cast = PyDict_GetItem(psyco_types, type))) {
                Dprintf("_pq_fetch_tuples: cast %d not found, using default",
                        PQftype(curs->pgres,i));
                cast = psyco_default_cast;
            }
        }
        /* else if we got binary tuples and if we got a field that
           is binary use the default cast
           FIXME: what the hell am I trying to do here? This just can't work..
        */
        else if (pgbintuples && cast == psyco_default_binary_cast) {
            Dprintf("_pq_fetch_tuples: Binary cursor and "
                    "binary field: %i using default cast",
                    PQftype(curs->pgres,i));
            cast = psyco_default_cast;
        }
        Dprintf("_pq_fetch_tuples: using cast at %p (%s) for type %d",
                cast, PyString_AS_STRING(((typecastObject*)cast)->name),
                PQftype(curs->pgres,i));
        Py_INCREF(cast);
        PyTuple_SET_ITEM(curs->casts, i, cast);
    

        /* 1/ fill the other fields */
        PyTuple_SET_ITEM(dtitem, 0,
                         PyString_FromString(PQfname(curs->pgres, i)));
        PyTuple_SET_ITEM(dtitem, 1, type);

        /* 2/ display size is the maximum size of this field result tuples. */
        if (dsize && dsize[i] >= 0) {
            PyTuple_SET_ITEM(dtitem, 2, PyInt_FromLong(dsize[i]));
        }
        else {
            Py_INCREF(Py_None);
            PyTuple_SET_ITEM(dtitem, 2, Py_None);
        }

        /* 3/ size on the backend */
        if (fmod > 0) fmod = fmod - sizeof(int);
        if (fsize == -1) {
            if (ftype == NUMERICOID) {
                PyTuple_SET_ITEM(dtitem, 3,
                                 PyInt_FromLong((fmod >> 16) & 0xFFFF));
            }
            else { /* If variable length record, return maximum size */
                PyTuple_SET_ITEM(dtitem, 3, PyInt_FromLong(fmod));
            }
        }
        else {
            PyTuple_SET_ITEM(dtitem, 3, PyInt_FromLong(fsize));
        }

        /* 4,5/ scale and precision */
        if (ftype == NUMERICOID) {
            PyTuple_SET_ITEM(dtitem, 4, PyInt_FromLong((fmod >> 16) & 0xFFFF));
            PyTuple_SET_ITEM(dtitem, 5, PyInt_FromLong((fmod & 0xFFFF) - 4));
        }
        else {
            Py_INCREF(Py_None);
            PyTuple_SET_ITEM(dtitem, 4, Py_None);
            Py_INCREF(Py_None);
            PyTuple_SET_ITEM(dtitem, 5, Py_None);
        }

        /* 6/ FIXME: null_ok??? */
        Py_INCREF(Py_None);
        PyTuple_SET_ITEM(dtitem, 6, Py_None);
    }
    
    if (dsize) PyMem_Free(dsize);
}

#ifdef HAVE_PQPROTOCOL3
static int
_pq_copy_in_v3(cursorObject *curs)
{
    /* COPY FROM implementation when protocol 3 is available: this function
       uses the new PQputCopyData() and can detect errors and set the correct
       exception */
    PyObject *o;
    int length = 0, error = 0;
    
    while (1) {
        o = PyObject_CallMethod(curs->copyfile, "read", "i", curs->copysize);
        if (!o || !PyString_Check(o) || (length = PyString_Size(o)) == -1) {
            error = 1;
        }
        if (length == 0 || error == 1) break;
        
        Py_BEGIN_ALLOW_THREADS;
        if (PQputCopyData(curs->conn->pgconn,
                          PyString_AS_STRING(o), length) == -1) {
            error = 2;
        }
        Py_END_ALLOW_THREADS;

        if (error == 2) break;
        
        Py_DECREF(o);
    }
    
    Py_XDECREF(o);
    
    if (error == 0 || error == 2)
        /* 0 means that the copy went well, 2 that there was an error on the
           backend: in both cases we'll get the error message from the
           PQresult */
        PQputCopyEnd(curs->conn->pgconn, NULL);
    else
        PQputCopyEnd(curs->conn->pgconn, "error during .read() call");

    /* and finally we grab the operation result from the backend */
    IFCLEARPGRES(curs->pgres);
    while ((curs->pgres = PQgetResult(curs->conn->pgconn)) != NULL) {
        if (PQresultStatus(curs->pgres) == PGRES_FATAL_ERROR)
            pq_raise(curs->conn, curs, NULL, NULL);
        IFCLEARPGRES(curs->pgres);
    }

    return 1;
}
#endif
static int
_pq_copy_in(cursorObject *curs)
{
    /* COPY FROM implementation when protocol 3 is not available: this
       function can't fail but the backend will send an ERROR notice that will
       be catched by our notice collector */
    PyObject *o;

    while (1) {
        o = PyObject_CallMethod(curs->copyfile, "readline", NULL);
        if (!o || o == Py_None || PyString_GET_SIZE(o) == 0) break;
        if (PQputline(curs->conn->pgconn, PyString_AS_STRING(o)) != 0) {
            Py_DECREF(o);
            return -1;
        }
        Py_DECREF(o);
    }
    Py_XDECREF(o);
    PQputline(curs->conn->pgconn, "\\.\n");
    PQendcopy(curs->conn->pgconn);

    /* if for some reason we're using a protocol 3 libpq to connect to a
       protocol 2 backend we still need to cycle on the result set */
    IFCLEARPGRES(curs->pgres);
    while ((curs->pgres = PQgetResult(curs->conn->pgconn)) != NULL) {
        if (PQresultStatus(curs->pgres) == PGRES_FATAL_ERROR)
            pq_raise(curs->conn, curs, NULL, NULL);
        IFCLEARPGRES(curs->pgres);
    }

    return 1;
}

#ifdef HAVE_PQPROTOCOL3
static int
_pq_copy_out_v3(cursorObject *curs)
{
    char *buffer;
    int len;
    
    while (1) {
        Py_BEGIN_ALLOW_THREADS;
        len = PQgetCopyData(curs->conn->pgconn, &buffer, 0);
        Py_END_ALLOW_THREADS;
            
        if (len > 0 && buffer) {
            PyObject_CallMethod(curs->copyfile, "write", "s#", buffer, len);
            PQfreemem(buffer);
        }
        /* we break on len == 0 but note that that should *not* happen,
           because we are not doing an async call (if it happens blame
           postgresql authors :/) */
        else if (len <= 0) break;
    }
    
    if (len == -2) {
        pq_raise(curs->conn, NULL, NULL, NULL);
        return -1;
    }

    /* and finally we grab the operation result from the backend */
    IFCLEARPGRES(curs->pgres);
    while ((curs->pgres = PQgetResult(curs->conn->pgconn)) != NULL) {
        if (PQresultStatus(curs->pgres) == PGRES_FATAL_ERROR)
            pq_raise(curs->conn, curs, NULL, NULL);
        IFCLEARPGRES(curs->pgres);
    }
    return 1;
}
#endif

static int
_pq_copy_out(cursorObject *curs)
{
    char buffer[4096];
    int status, len;

    while (1) {
        Py_BEGIN_ALLOW_THREADS;
        status = PQgetline(curs->conn->pgconn, buffer, 4096);
        Py_END_ALLOW_THREADS;
        if (status == 0) {
            if (buffer[0] == '\\' && buffer[1] == '.') break;

            len = strlen(buffer);
            buffer[len++] = '\n';
        }
        else if (status == 1) {
            len = 4096-1;
        }
        else {
            return -1;
        }
        
        PyObject_CallMethod(curs->copyfile, "write", "s#", buffer, len);
    }

    status = 1;
    if (PQendcopy(curs->conn->pgconn) != 0)
        status = -1;
    
    /* if for some reason we're using a protocol 3 libpq to connect to a
       protocol 2 backend we still need to cycle on the result set */
    IFCLEARPGRES(curs->pgres);
    while ((curs->pgres = PQgetResult(curs->conn->pgconn)) != NULL) {
        if (PQresultStatus(curs->pgres) == PGRES_FATAL_ERROR)
            pq_raise(curs->conn, curs, NULL, NULL);
        IFCLEARPGRES(curs->pgres);
    }

    return status;
}

int
pq_fetch(cursorObject *curs)
{
    int pgstatus, ex = -1;

    /* even if we fail, we remove any information about the previous query */
    curs_reset(curs);
    
    /* we check the result from the previous execute; if the result is not
       already there, we need to consume some input and go to sleep until we
       get something edible to eat */
    if (!curs->pgres) {
        
        Dprintf("pq_fetch: no data: entering polling loop");
        
        while (pq_is_busy(curs->conn) > 0) {
            fd_set rfds;
            struct timeval tv;
            int sval, sock;

            Py_BEGIN_ALLOW_THREADS;
            pthread_mutex_lock(&(curs->conn->lock));

            sock = PQsocket(curs->conn->pgconn);
            FD_ZERO(&rfds);
            FD_SET(sock, &rfds);

            /* set a default timeout of 5 seconds
               TODO: make use of the timeout, maybe allowing the user to
               make a non-blocking (timeouted) call to fetchXXX */
            tv.tv_sec = 5;
            tv.tv_usec = 0;

            Dprintf("pq_fetch: entering PDflush() loop");
            while (PQflush(curs->conn->pgconn) != 0);
            sval = select(sock+1, &rfds, NULL, NULL, &tv);

            pthread_mutex_unlock(&(curs->conn->lock));
            Py_END_ALLOW_THREADS;
        }

        Dprintf("pq_fetch: data is probably ready");
        IFCLEARPGRES(curs->pgres);
        curs->pgres = PQgetResult(curs->conn->pgconn);
    }

    /* check for PGRES_FATAL_ERROR result */
    /* FIXME: I am not sure we need to check for critical error here.
    if (curs->pgres == NULL) {
        Dprintf("pq_fetch: got a NULL pgres, checking for critical");
        pq_set_critical(curs->conn);
        if (curs->conn->critical) {
            pq_resolve_critical(curs->conn);
            return -1;
        }
        else {
            return 0;
        }
    }
    */
    
    if (curs->pgres == NULL) return 0;
    
    pgstatus = PQresultStatus(curs->pgres);
    Dprintf("pq_fetch: pgstatus = %s", PQresStatus(pgstatus));

    /* backend status message */
    Py_XDECREF(curs->pgstatus);
    curs->pgstatus = PyString_FromString(PQcmdStatus(curs->pgres));

    switch(pgstatus) {

    case PGRES_COMMAND_OK:
        Dprintf("pq_fetch: command returned OK (no tuples)");
        curs->rowcount = atoi(PQcmdTuples(curs->pgres));
        curs->lastoid = PQoidValue(curs->pgres);
        CLEARPGRES(curs->pgres);
        ex = 1;
        break;

    case PGRES_COPY_OUT:
        Dprintf("pq_fetch: data from a COPY TO (no tuples)");
#ifdef HAVE_PQPROTOCOL3
        if (curs->conn->protocol == 3)
            ex = _pq_copy_out_v3(curs);
        else
#endif
            ex = _pq_copy_out(curs);
        curs->rowcount = -1;
        /* error caught by out glorious notice handler */
        if (PyErr_Occurred()) ex = -1;
        IFCLEARPGRES(curs->pgres);
        break;
        
    case PGRES_COPY_IN:
        Dprintf("pq_fetch: data from a COPY FROM (no tuples)");
#ifdef HAVE_PQPROTOCOL3
        if (curs->conn->protocol == 3)        
            ex = _pq_copy_in_v3(curs);
        else
#endif
            ex = _pq_copy_in(curs);
        curs->rowcount = -1;
        /* error caught by out glorious notice handler */
        if (PyErr_Occurred()) ex = -1;
        IFCLEARPGRES(curs->pgres);
        break;
        
    case PGRES_TUPLES_OK:
        Dprintf("pq_fetch: data from a SELECT (got tuples)");
        curs->rowcount = PQntuples(curs->pgres);
        _pq_fetch_tuples(curs); ex = 0;
        /* don't clear curs->pgres, because it contains the results! */
        break;
        
    default:
        Dprintf("pq_fetch: uh-oh, something FAILED");
        pq_raise(curs->conn, curs, NULL, NULL);
        IFCLEARPGRES(curs->pgres);
        ex = -1;
        break;
    }

    Dprintf("pq_fetch: fetching done; check for critical errors");
    
    /* error checking, close the connection if necessary (some critical errors
       are not really critical, like a COPY FROM error: if that's the case we
       raise the exception but we avoid to close the connection) */
    if (curs->conn->critical) {
        if (ex == -1) {
            pq_resolve_critical(curs->conn, 1);
        }
        else {
            pq_resolve_critical(curs->conn, 0);
        }
        return -1;
    }
    
    return ex;
}