psycopg/psycopg2
1
1
Fork 0
mirror of https://github.com/psycopg/psycopg2.git synced 2025-01-31 09:24:07 +03:00
Code Issues Packages Projects Releases Wiki Activity

Added autocommit property on connection

Browse Source
This commit is contained in:
Daniele Varrazzo 2011-06-02 01:16:22 +01:00
parent ea03ffbf76
commit 389f2cf1d0
5 changed files with 143 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
psycopg/connection.h
View File

@ -119,6 +119,8 @@ typedef struct {
int equote; /* use E''-style quotes for escaped strings */
PyObject *weakreflist; /* list of weak references */
int autocommit;
} connectionObject;
/* C-callable functions in connection_int.c and connection_ext.c */

3
psycopg/connection_int.c
View File

@ -989,7 +989,8 @@ conn_set(connectionObject *self, const char *param, const char *value)
int
conn_set_autocommit(connectionObject *self, int value)
{
return -1;
self->autocommit = value;
return 0;
}
/* conn_switch_isolation_level - switch isolation level on the connection */

42
psycopg/connection_type.c
View File

@ -517,6 +517,42 @@ psyco_conn_set_transaction(connectionObject *self, PyObject *args, PyObject *kwa
}
#define psyco_conn_autocommit_doc \
"set or return the autocommit status."
static PyObject *
psyco_conn_autocommit_get(connectionObject *self)
{
PyObject *ret;
ret = self->autocommit ? Py_True : Py_False;
Py_INCREF(ret);
return ret;
}
static PyObject *
_psyco_conn_autocommit_set_checks(connectionObject *self)
{
/* wrapper to use the EXC_IF macros.
* return NULL in case of error, else whatever */
EXC_IF_CONN_CLOSED(self);
EXC_IF_CONN_ASYNC(self, autocommit);
EXC_IF_IN_TRANSACTION(self, autocommit);
return Py_None; /* borrowed */
}
static int
psyco_conn_autocommit_set(connectionObject *self, PyObject *pyvalue)
{
int value;
if (!_psyco_conn_autocommit_set_checks(self)) { return -1; }
if (-1 == (value = PyObject_IsTrue(pyvalue))) { return -1; }
if (0 != conn_set_autocommit(self, value)) { return -1; }
return 0;
}
/* set_isolation_level method - switch connection isolation level */
#define psyco_conn_set_isolation_level_doc \
@ -927,6 +963,12 @@ static struct PyGetSetDef connectionObject_getsets[] = {
EXCEPTION_GETTER(IntegrityError),
EXCEPTION_GETTER(DataError),
EXCEPTION_GETTER(NotSupportedError),
#ifdef PSYCOPG_EXTENSIONS
{ "autocommit",
(getter)psyco_conn_autocommit_get,
(setter)psyco_conn_autocommit_set,
psyco_conn_autocommit_doc },
#endif
{NULL}
};
#undef EXCEPTION_GETTER

5
psycopg/pqpath.c
View File

@ -417,6 +417,11 @@ pq_begin_locked(connectionObject *conn, PGresult **pgres, char **error,
Dprintf("pq_begin_locked: pgconn = %p, isolevel = %ld, status = %d",
conn->pgconn, conn->isolation_level, conn->status);
if (conn->autocommit) {
Dprintf("pq_begin_locked: autocommit");
return 0;
}
if (conn->isolation_level == ISOLATION_LEVEL_AUTOCOMMIT
|| conn->status != CONN_STATUS_READY) {
Dprintf("pq_begin_locked: transaction in progress");

92
tests/test_connection.py
View File

@ -851,6 +851,98 @@ class TransactionControlTests(unittest.TestCase):
self.conn.set_transaction, readonly=True, deferrable=True)
class AutocommitTests(unittest.TestCase):
def setUp(self):
self.conn = psycopg2.connect(dsn)
def tearDown(self):
if not self.conn.closed:
self.conn.close()
def test_default_no_autocommit(self):
self.assert_(not self.conn.autocommit)
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_IDLE)
cur = self.conn.cursor()
cur.execute('select 1;')
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_BEGIN)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_INTRANS)
self.conn.rollback()
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_IDLE)
def test_set_autocommit(self):
self.conn.autocommit = True
self.assert_(self.conn.autocommit)
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_IDLE)
cur = self.conn.cursor()
cur.execute('select 1;')
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_IDLE)
self.conn.autocommit = False
self.assert_(not self.conn.autocommit)
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_IDLE)
cur.execute('select 1;')
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_BEGIN)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_INTRANS)
def test_set_intrans_error(self):
cur = self.conn.cursor()
cur.execute('select 1;')
self.assertRaises(psycopg2.ProgrammingError,
setattr, self.conn, 'autocommit', True)
def test_set_transaction_autocommit(self):
self.conn.set_transaction(autocommit=True)
self.assert_(self.conn.autocommit)
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_IDLE)
cur = self.conn.cursor()
cur.execute('select 1;')
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_IDLE)
self.conn.set_transaction(autocommit=False)
self.assert_(not self.conn.autocommit)
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_IDLE)
cur.execute('select 1;')
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_BEGIN)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_INTRANS)
self.conn.rollback()
self.conn.set_transaction('serializable', readonly=True, autocommit=True)
self.assert_(self.conn.autocommit)
cur.execute('select 1;')
self.assertEqual(self.conn.status, psycopg2.extensions.STATUS_READY)
self.assertEqual(self.conn.get_transaction_status(),
psycopg2.extensions.TRANSACTION_STATUS_IDLE)
cur.execute("SHOW default_transaction_isolation;")
self.assertEqual(cur.fetchone()[0], 'serializable')
cur.execute("SHOW default_transaction_read_only;")
self.assertEqual(cur.fetchone()[0], 'on')
def test_suite():
return unittest.TestLoader().loadTestsFromName(__name__)