My favorites | Sign in
Project Home Downloads Wiki
READ-ONLY: This project has been archived. For more information see this post.
Search
for
  Advanced search   Search tips   Subscriptions

Issue 82 attachment: test_with_pyodbc.py (11.6 KB)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
Test for pyodbc failing when used with *with_statement* and executing
SQL statement not on one row.

Other methods (test_simple1_works and test_simple2_works) are shown only
for information purpose. The mind-blowing is the difference between:
- test_with_works and
- test_with_fails
given the the only difference between two methods is that in the second one
the cursor is assigned to a temp variable first.

----
RESULTS of runs (must run one method at a time)
----

INFO:test_simple1_works:Inserting row 0 ...
INFO:test_simple1_works:Inserted row 0 ...
INFO:test_simple1_works:Inserting row 1 ...
Traceback (most recent call last):
File "C:\AltBeta\Dev\hg-ivanoe-pg\VariousTests\test_with_pyodbc.py", line 219, in <module>
print t.test_simple1_works() # this fails with exception
File "C:\AltBeta\Dev\hg-ivanoe-pg\VariousTests\test_with_pyodbc.py", line 176, in test_simple1_works
db_conn.cursor().execute(Test_pyodbc_connect._SQL_INSERT, (1,))
pyodbc.IntegrityError: ('23000', "[23000] [Microsoft][SQL Server Native Client 10.0][SQL Server]Cannot insert duplicate key row in object 'dbo.test_pyodbc_table' with unique index 'test_pyodbc_table_SomeUniqueValue_u_nci'. (2601) (SQLExecDirectW); [01000] [Microsoft][SQL Server Native Client 10.0][SQL Server]The statement has been terminated. (3621)")

----

INFO:test_simple2_works:Inserting row 0 ...
INFO:test_simple2_works:Inserted row 0 ...
INFO:test_simple2_works:Inserting row 1 ...
Traceback (most recent call last):
File "C:\AltBeta\Dev\hg-ivanoe-pg\VariousTests\test_with_pyodbc.py", line 223, in <module>
print t.test_simple2_works() # this fails with exception
File "C:\AltBeta\Dev\hg-ivanoe-pg\VariousTests\test_with_pyodbc.py", line 191, in test_simple2_works
curs.execute(Test_pyodbc_connect._SQL_INSERT, (1,))
pyodbc.IntegrityError: ('23000', "[23000] [Microsoft][SQL Server Native Client 10.0][SQL Server]Cannot insert duplicate key row in object 'dbo.test_pyodbc_table' with unique index 'test_pyodbc_table_SomeUniqueValue_u_nci'. (2601) (SQLExecDirectW); [01000] [Microsoft][SQL Server Native Client 10.0][SQL Server]The statement has been terminated. (3621)")

----

DEBUG:SQL:Block started, connecting to "DRIVER={SQL Server Native Client 10.0};SERVER=test_srv;DATABASE=tempdb;Trusted_Connection=Yes"
DEBUG:SQL:Block started, connected to "DRIVER={SQL Server Native Client 10.0};SERVER=test_srv;DATABASE=tempdb;Trusted_Connection=Yes"
INFO:test_with_works:Inserting row 0 ...
INFO:test_with_works:Inserted row 0 ...
INFO:test_with_works:Inserting row 1 ...
ERROR:SQL:Error within block, rolling back and re-raising: <class 'pyodbc.IntegrityError'> (('23000', "[23000] [Microsoft][SQL Server Native Client 10.0][SQL Server]Cannot insert duplicate key row in object 'dbo.test_pyodbc_table' with unique index 'test_pyodbc_table_SomeUniqueValue_u_nci'. (2601) (SQLExecDirectW); [01000] [Microsoft][SQL Server Native Client 10.0][SQL Server]The statement has been terminated. (3621)"))
DEBUG:SQL:Block finished, closing...
DEBUG:SQL:Block finished, closed.
Traceback (most recent call last):
File "C:\AltBeta\Dev\hg-ivanoe-pg\VariousTests\test_with_pyodbc.py", line 226, in <module>
print t.test_with_works() # this fails with exception re-raised
File "C:\AltBeta\Dev\hg-ivanoe-pg\VariousTests\test_with_pyodbc.py", line 203, in test_with_works
db_conn.cursor().execute(Test_pyodbc_connect._SQL_INSERT, (1,))
pyodbc.IntegrityError: ('23000', "[23000] [Microsoft][SQL Server Native Client 10.0][SQL Server]Cannot insert duplicate key row in object 'dbo.test_pyodbc_table' with unique index 'test_pyodbc_table_SomeUniqueValue_u_nci'. (2601) (SQLExecDirectW); [01000] [Microsoft][SQL Server Native Client 10.0][SQL Server]The statement has been terminated. (3621)")

----

DEBUG:SQL:Block started, connecting to "DRIVER={SQL Server Native Client 10.0};SERVER=test_srv;DATABASE=tempdb;Trusted_Connection=Yes"
DEBUG:SQL:Block started, connected to "DRIVER={SQL Server Native Client 10.0};SERVER=test_srv;DATABASE=tempdb;Trusted_Connection=Yes"
INFO:test_with_fails:Inserting row 0 ...
INFO:test_with_fails:Inserted row 0 ...
INFO:test_with_fails:Inserting row 1 ...
ERROR:SQL:Error within block, rolling back and re-raising: <class 'pyodbc.IntegrityError'> (('23000', "[23000] [Microsoft][SQL Server Native Client 10.0][SQL Server]Cannot insert duplicate key row in object 'dbo.test_pyodbc_table' with unique index 'test_pyodbc_table_SomeUniqueValue_u_nci'. (2601) (SQLExecDirectW); [01000] [Microsoft][SQL Server Native Client 10.0][SQL Server]The statement has been terminated. (3621)"))
DEBUG:SQL:Block finished, closing...
DEBUG:SQL:Block finished, closed.
Traceback (most recent call last):
File "C:\AltBeta\Dev\hg-ivanoe-pg\VariousTests\test_with_pyodbc.py", line 221, in <module>
print t.test_with_fails() # this is the bad one: win32 error
File "C:\AltBeta\Dev\hg-ivanoe-pg\VariousTests\test_with_pyodbc.py", line 210, in test_with_fails
curs.execute(Test_pyodbc_connect._SQL_INSERT, (1,))
pyodbc.IntegrityError: ('23000', "[23000] [Microsoft][SQL Server Native Client 10.0][SQL Server]Cannot insert duplicate key row in object 'dbo.test_pyodbc_table' with unique index 'test_pyodbc_table_SomeUniqueValue_u_nci'. (2601) (SQLExecDirectW); [01000] [Microsoft][SQL Server Native Client 10.0][SQL Server]The statement has been terminated. (3621)")

... and the error:
---------------------------
Visual Studio Just-In-Time Debugger
---------------------------
An unhandled win32 exception occurred in python.exe [5080]. Just-In-Time debugging this exception failed with the following error: No installed debugger has Just-In-Time debugging enabled. In Visual Studio, Just-In-Time debugging can be enabled from Tools/Options/Debugging/Just-In-Time.

Check the documentation index for 'Just-in-time debugging, errors' for more information.
---------------------------
OK
---------------------------
"""

from __future__ import with_statement
from contextlib import contextmanager

import logging
import sys
import pyodbc

@contextmanager
def pyodbc_connect(db_conn_str, autocommit=False):
"""
This method encapsulates pyodbc.connect in such a way that it
can be used in 'with' statement. It takes care to rollback transaction
when an exception occurs, and to commit otherwise.
Also the connection is closed on __exit__ from 'with'

@param db_conn_str: connection string
@type db_conn_str: str
@param autocommit: parameter to pass to pyodbc.connect
@type autocommit: bool

Usage example:
with pyodbc_connect(db_connection_string) as db_conn:
print db_conn # prints something like "<pyodbc.Connection object at 0x00A91780>"
r = db_conn.execute("SELECT * FROM MyTable")
#raise BaseException, "TestException" # if uncommented, will rollback, exit with, and close the connection
d = r.fetchone()
# at this point any uncommitted changes are committed and connection is closed
"""
logger = logging.getLogger('SQL')
conn = None
try:
logger.debug('Block started, connecting to "%s"' % db_conn_str)
conn = pyodbc.connect(db_conn_str, autocommit=autocommit)
logger.debug('Block started, connected to "%s"' % db_conn_str)
yield conn
except:
logger.error('Error within block, rolling back and re-raising: %s (%s)',
sys.exc_info()[0], sys.exc_info()[1])
if conn:
conn.rollback()
raise
else:
logger.debug('Block finished, committing...')
if conn:
conn.commit()
logger.debug('Block finished, committed.')
finally:
logger.debug('Block finished, closing...')
if conn:
conn.close()
logger.debug('Block finished, closed.')


class Test_pyodbc_connect():

_SQL_CREATE_TABLE = """
IF EXISTS (SELECT id FROM dbo.sysobjects WHERE id = OBJECT_ID(N'"test_pyodbc_table"') AND OBJECTPROPERTY(id, N'IsUserTable') = 1)
DROP TABLE "test_pyodbc_table";

CREATE TABLE "test_pyodbc_table" (
"ID" INTEGER IDENTITY NOT NULL,
"SomeUniqueValue" INTEGER NOT NULL,

CONSTRAINT "test_pyodbc_table_PK" PRIMARY KEY ("ID")
);

CREATE UNIQUE NONCLUSTERED INDEX "test_pyodbc_table_SomeUniqueValue_u_nci"
ON "test_pyodbc_table" (
"SomeUniqueValue" ASC
)
"""

_SQL_INSERT = """INSERT INTO test_pyodbc_table (SomeUniqueValue) VALUES (?)"""

def __init__(self):
self._logger = logging.getLogger()
self._logger.setLevel(logging.DEBUG)
self._db_conn_str = """DRIVER={SQL Server Native Client 10.0};SERVER=test_srv;DATABASE=tempdb;Trusted_Connection=Yes"""

# create the table
_conn = pyodbc.connect(self._db_conn_str)
_conn.cursor().execute(Test_pyodbc_connect._SQL_CREATE_TABLE)
_conn.commit()
_conn.close()

def __del__(self):
# delete the table
_SQL_DELETE_TABLE = """
IF EXISTS (SELECT id FROM dbo.sysobjects WHERE id = OBJECT_ID(N'"test_pyodbc_table"') AND OBJECTPROPERTY(id, N'IsUserTable') = 1)
DROP TABLE "test_pyodbc_table";
"""
_conn = pyodbc.connect(self._db_conn_str)
_conn.cursor().execute(_SQL_DELETE_TABLE)
_conn.commit()
_conn.close()

def test_simple1_works(self):
""" Use pyodbc.connect() and immediatelly execute sql on cursor. """
logger = logging.getLogger('test_simple1_works')
db_conn = pyodbc.connect(self._db_conn_str)
if db_conn:
for row in range(2):
logger.info("Inserting row %d ...", row)
db_conn.cursor().execute(Test_pyodbc_connect._SQL_INSERT, (1,))
logger.info("Inserted row %d ...", row)
return True

def test_simple2_works(self):
""" Use pyodbc.connect() and execute sql on cursor variable. """
logger = logging.getLogger('test_simple2_works')
db_conn = pyodbc.connect(self._db_conn_str)
if db_conn:
for row in range(2):
logger.info("Inserting row %d ...", row)
curs = db_conn.cursor()
curs.execute(Test_pyodbc_connect._SQL_INSERT, (1,))
logger.info("Inserted row %d ...", row)
return True

def test_with_works(self):
""" Use contextmanager and immediatelly execute sql on cursor. """
logger = logging.getLogger('test_with_works')
with pyodbc_connect(self._db_conn_str) as db_conn:
for row in range(2):
logger.info("Inserting row %d ...", row)
db_conn.cursor().execute(Test_pyodbc_connect._SQL_INSERT, (1,))
logger.info("Inserted row %d ...", row)
return True

def test_with_fails(self):
""" Use contextmanager and execute sql on cursor variable.
FAILURE: produces win32 error (Just-In-Time debugger)
"""
logger = logging.getLogger('test_with_fails')
with pyodbc_connect(self._db_conn_str) as db_conn:
for row in range(2):
logger.info("Inserting row %d ...", row)
curs = db_conn.cursor()
curs.execute(Test_pyodbc_connect._SQL_INSERT, (1,))
logger.info("Inserted row %d ...", row)
return True


logging.basicConfig()

t = Test_pyodbc_connect()
#t.test_simple1_works() # this fails with exception
#t.test_simple2_works() # this fails with exception
#t.test_with_works() # this fails with exception re-raised
t.test_with_fails() # this is the bad one: win32 error
Powered by Google Project Hosting