Created
July 6, 2012 17:08
-
-
Save niwinz/3061376 to your computer and use it in GitHub Desktop.
Sqlite3 isolation_level and savepoint suport test
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
[3/4.3.17]niwi@vaio:~> python2 savepoint_sqlite3_test.py | |
python version: 2.7.3 (default, Apr 24 2012, 00:00:54) | |
[GCC 4.7.0 20120414 (prerelease)] | |
PySQLite version: 2.6.0 | |
sqlite3 version: 3.7.13 | |
Test with isolation_level = None (autocommit mode) | |
********************************************************************** | |
ROW: (0, 200) | |
ROW: (5, 205) | |
Test with isolation_level = (default) and use_savepoint = False | |
********************************************************************** | |
Traceback (most recent call last): | |
File "savepoint_sqlite3_test.py", line 63, in <module> | |
for row in with_isolation_level(use_savepoint=False, isolation_level=""): | |
File "savepoint_sqlite3_test.py", line 47, in with_isolation_level | |
cur.execute("COMMIT") | |
OperationalError: cannot commit - no transaction is active | |
Test with isolation_level = (default) and use_savepoint = False | |
********************************************************************** | |
Traceback (most recent call last): | |
File "savepoint_sqlite3_test.py", line 72, in <module> | |
for row in with_isolation_level(use_savepoint=True, isolation_level=""): | |
File "savepoint_sqlite3_test.py", line 33, in with_isolation_level | |
cur.execute("rollback to savepoint spTest;") | |
OperationalError: no such savepoint: spTest |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# -*- coding: utf-8 -*- | |
import sys | |
import sqlite3 | |
import traceback | |
print 'python version:', sys.version | |
print 'PySQLite version:', sqlite3.version | |
print 'sqlite3 version:', sqlite3.sqlite_version | |
def with_isolation_level(isolation_level="", use_savepoint=False): | |
conn = sqlite3.connect(':memory:') | |
conn.isolation_level = isolation_level | |
cur = conn.cursor() | |
cur.execute("create table example (A, B);") | |
# With isolation_level != None this is redundant, because | |
# sqlite3 module executes this implicitly, and sqlite3 ignores this. | |
cur.execute("BEGIN") | |
cur.execute("insert into example values (?, ?);", (0,200)) | |
if use_savepoint: | |
cur.execute("savepoint spTest;") | |
cur.execute("insert into example values (?, ?);", (1,201)) | |
cur.execute("insert into example values (?, ?);", (2,202)) | |
if use_savepoint: | |
# see COMMIT commet | |
cur.execute("rollback to savepoint spTest;") | |
cur.execute("insert into example values (?, ?);", (5,205)) | |
# On isolation_level != None, the explicitly BEGIN statements are ignored | |
# and on intent execute this, raises: "OperationalError: cannot commit - no transaction is active" | |
# beacause the explicitly BEGIN transaction are not tracked. | |
# ```By default, the sqlite3 module opens transactions implicitly before a Data Modification Language | |
# (DML) statement (i.e. INSERT/UPDATE/DELETE/REPLACE), and commits transactions implicitly before a | |
# non-DML, non-query statement (i. e. anything other than SELECT or the aforementioned).``` | |
# SOURCE: http://docs.python.org/library/sqlite3.html#controlling-transactions | |
# Same behavior is for savepoints, only works on isolation_level == None | |
cur.execute("COMMIT") | |
cur.execute("select * from example;") | |
return list(cur) | |
print "\nTest with isolation_level = None (autocommit mode)" | |
print "*"*70 | |
for row in with_isolation_level(use_savepoint=True, isolation_level=None): | |
print "ROW:", row | |
print "\nTest with isolation_level = "" (default) and use_savepoint = False" | |
print "*"*70 | |
try: | |
for row in with_isolation_level(use_savepoint=False, isolation_level=""): | |
print "ROW:", row | |
except Exception: | |
traceback.print_exc() | |
print "\nTest with isolation_level = "" (default) and use_savepoint = False" | |
print "*"*70 | |
try: | |
for row in with_isolation_level(use_savepoint=True, isolation_level=""): | |
print "ROW:", row | |
except Exception: | |
traceback.print_exc() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment