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
|
from sqlalchemy import util
from sqlalchemy.engine import base
"""Provides a thread-local transactional wrapper around the root Engine class.
Provides begin/commit methods on the engine itself which correspond to
a thread-local transaction.
"""
class TLSession(object):
def __init__(self, engine):
self.engine = engine
self.__tcount = 0
def get_connection(self, close_with_result=False):
try:
return self.__transaction._increment_connect()
except AttributeError:
return TLConnection(self, self.engine.pool.connect(), close_with_result=close_with_result)
def reset(self):
try:
self.__transaction._force_close()
del self.__transaction
del self.__trans
except AttributeError:
pass
self.__tcount = 0
def _conn_closed(self):
if self.__tcount == 1:
self.__trans._trans.rollback()
self.reset()
def in_transaction(self):
return self.__tcount > 0
def prepare(self):
if self.__tcount == 1:
try:
self.__trans._trans.prepare()
finally:
self.reset()
def begin_twophase(self, xid=None):
raise NotImplementedError(
"Two phase transactions not yet implemented for 'threadlocal' "
"strategy")
def _dont_begin_twophase(self, xid=None):
if self.__tcount == 0:
self.__transaction = self.get_connection()
self.__trans = self.__transaction._begin_twophase(xid=xid)
self.__tcount += 1
return self.__trans
def begin(self, **kwargs):
if self.__tcount == 0:
self.__transaction = self.get_connection()
self.__trans = self.__transaction._begin(**kwargs)
self.__tcount += 1
return self.__trans
def rollback(self):
if self.__tcount > 0:
try:
self.__trans._trans.rollback()
finally:
self.reset()
def commit(self):
if self.__tcount == 1:
try:
self.__trans._trans.commit()
finally:
self.reset()
elif self.__tcount > 1:
self.__tcount -= 1
def is_begun(self):
return self.__tcount > 0
class TLConnection(base.Connection):
def __init__(self, session, connection, close_with_result):
base.Connection.__init__(self, session.engine, connection, close_with_result=close_with_result)
self.__session = session
self.__opencount = 1
session = property(lambda s:s.__session)
def _increment_connect(self):
self.__opencount += 1
return self
def _begin(self, **kwargs):
return TLTransaction(
super(TLConnection, self).begin(**kwargs), self.__session)
def _begin_twophase(self, xid=None):
return TLTransaction(
super(TLConnection, self).begin_twophase(xid=xid), self.__session)
def in_transaction(self):
return self.session.in_transaction()
def begin(self, **kwargs):
return self.session.begin(**kwargs)
def begin_twophase(self, xid=None):
return self.session.begin_twophase(xid=xid)
def close(self):
if self.__opencount == 1:
base.Connection.close(self)
self.__session._conn_closed()
self.__opencount -= 1
def _force_close(self):
self.__opencount = 0
base.Connection.close(self)
class TLTransaction(base.Transaction):
def __init__(self, trans, session):
self._trans = trans
self._session = session
connection = property(lambda s:s._trans.connection)
is_active = property(lambda s:s._trans.is_active)
def rollback(self):
self._session.rollback()
def prepare(self):
self._session.prepare()
def commit(self):
self._session.commit()
def __enter__(self):
return self
def __exit__(self, type, value, traceback):
self._trans.__exit__(type, value, traceback)
class TLEngine(base.Engine):
"""An Engine that includes support for thread-local managed transactions.
The TLEngine relies upon its Pool having "threadlocal" behavior,
so that once a connection is checked out for the current thread,
you get that same connection repeatedly.
"""
def __init__(self, *args, **kwargs):
"""Construct a new TLEngine."""
super(TLEngine, self).__init__(*args, **kwargs)
self.context = util.ThreadLocal()
def _session(self):
if not hasattr(self.context, 'session'):
self.context.session = TLSession(self)
return self.context.session
session = property(_session, doc="Returns the current thread's TLSession")
def contextual_connect(self, **kwargs):
"""Return a TLConnection which is thread-locally scoped."""
return self.session.get_connection(**kwargs)
def begin(self, **kwargs):
return self.session.begin(**kwargs)
def commit(self):
self.session.commit()
def rollback(self):
self.session.rollback()
def __repr__(self):
return 'TLEngine(%s)' % str(self.url)
|
