Skip to content

Commit 5ae09dd

Browse files
committed
fix(sqlserver): fix transaction level tracking, cursor await, and execute options
- Maintain _transaction_level via base begin/commit/rollback in sync and async transaction managers; quote savepoint names with format_identifier - Make AsyncSQLServerTransactionManager inherit AsyncTransactionManager - Await aioodbc connection.cursor() in _get_cursor so callers receive a real Cursor instead of the _ContextManager wrapper - Merge column_mapping/column_adapters kwargs into ExecutionOptions in execute() (sync and async) - Populate last_insert_id for INSERT via @@IDENTITY (pyodbc exposes no lastrowid and SCOPE_IDENTITY is reset across batches)
1 parent 30ab44c commit 5ae09dd

5 files changed

Lines changed: 204 additions & 136 deletions

File tree

‎src/rhosocial/activerecord/backend/impl/sqlserver/async_backend.py‎

Lines changed: 33 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -183,7 +183,16 @@ async def _get_cursor(self):
183183
self.log(logging.DEBUG, "No connection, connecting...")
184184
await self.connect()
185185

186-
return self._connection.cursor()
186+
return await self._connection.cursor()
187+
188+
async def _get_scope_identity_async(self) -> Optional[int]:
189+
"""Get the identity value of the last INSERT via @@IDENTITY."""
190+
async with await self._get_cursor() as cursor:
191+
await cursor.execute("SELECT @@IDENTITY")
192+
row = await cursor.fetchone()
193+
if row and row[0] is not None:
194+
return int(row[0])
195+
return None
187196

188197
async def execute(
189198
self,
@@ -205,17 +214,30 @@ async def execute(
205214
QueryResult containing the query results
206215
"""
207216
from rhosocial.activerecord.backend.options import ExecutionOptions, StatementType
208-
217+
218+
sql_upper = sql.strip().upper()
219+
209220
if options is None:
210-
sql_upper = sql.strip().upper()
211221
if sql_upper.startswith(("SELECT", "WITH", "EXEC", "EXECUTE")):
212222
stmt_type = StatementType.DQL
213223
elif sql_upper.startswith(("INSERT", "UPDATE", "DELETE", "MERGE")):
214224
stmt_type = StatementType.DML
215225
else:
216226
stmt_type = StatementType.DDL
217-
218-
options = ExecutionOptions(stmt_type=stmt_type)
227+
228+
column_mapping = kwargs.get("column_mapping")
229+
column_adapters = kwargs.get("column_adapters")
230+
231+
options = ExecutionOptions(
232+
stmt_type=stmt_type,
233+
column_adapters=column_adapters,
234+
column_mapping=column_mapping,
235+
)
236+
else:
237+
if "column_mapping" in kwargs:
238+
options.column_mapping = kwargs["column_mapping"]
239+
if "column_adapters" in kwargs:
240+
options.column_adapters = kwargs["column_adapters"]
219241

220242
if not self._connection:
221243
await self.connect()
@@ -276,6 +298,12 @@ async def execute(
276298
)
277299

278300
result = self._build_query_result(cursor, data, duration)
301+
302+
if options.stmt_type == StatementType.DML and sql_upper.startswith("INSERT"):
303+
scope_id = await self._get_scope_identity_async()
304+
if scope_id is not None:
305+
result.last_insert_id = scope_id
306+
279307
return result
280308

281309
except Exception as e:

‎src/rhosocial/activerecord/backend/impl/sqlserver/async_transaction.py‎

Lines changed: 75 additions & 86 deletions
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,10 @@
44
"""
55

66
import logging
7-
from contextlib import asynccontextmanager
8-
from typing import AsyncGenerator, List, Optional
7+
from typing import List, Optional
98

109
from rhosocial.activerecord.backend.transaction import (
10+
AsyncTransactionManager,
1111
TransactionState,
1212
IsolationLevel,
1313
TransactionMode,
@@ -19,40 +19,21 @@
1919
from .transaction import SQLServerIsolationLevel, ISOLATION_LEVEL_MAP
2020

2121

22-
class AsyncSQLServerTransactionManager:
22+
class AsyncSQLServerTransactionManager(AsyncTransactionManager):
2323
"""Asynchronous SQL Server transaction manager.
2424
2525
Provides async transaction management with savepoint support.
26+
Inherits the level tracking, nesting (savepoint-based) and owner-task
27+
bookkeeping from the core :class:`AsyncTransactionManager`, while the
28+
``_do_*`` layer emits SQL Server-specific statements.
2629
"""
2730

2831
def __init__(self, backend, logger=None):
29-
self._backend = backend
30-
self._logger = logger
31-
self._state = TransactionState.INACTIVE
32+
super().__init__(backend, logger)
3233
self._savepoints: List[str] = []
3334
self._sqlserver_isolation_level: Optional[SQLServerIsolationLevel] = None
3435
self._name: Optional[str] = None
3536

36-
@property
37-
def is_active(self) -> bool:
38-
"""Check if a transaction is currently active."""
39-
return self._state == TransactionState.ACTIVE
40-
41-
@property
42-
def state(self) -> TransactionState:
43-
"""Get the current transaction state."""
44-
return self._state
45-
46-
@property
47-
def isolation_level(self) -> Optional[IsolationLevel]:
48-
"""Get the current isolation level."""
49-
return self._isolation_level
50-
51-
@isolation_level.setter
52-
def isolation_level(self, level: Optional[IsolationLevel]):
53-
"""Set the isolation level for the next transaction."""
54-
self._isolation_level = level
55-
5637
async def begin(
5738
self,
5839
isolation_level: Optional[IsolationLevel] = None,
@@ -69,9 +50,6 @@ async def begin(
6950
name: Optional transaction name
7051
snapshot: If True, use SNAPSHOT isolation level
7152
"""
72-
if self._state == TransactionState.ACTIVE:
73-
raise TransactionError("Transaction already active")
74-
7553
if mode == TransactionMode.READ_ONLY:
7654
raise UnsupportedTransactionModeError(
7755
feature="READ ONLY transactions",
@@ -90,16 +68,25 @@ async def begin(
9068
self._sqlserver_isolation_level = SQLServerIsolationLevel.SNAPSHOT
9169
elif effective_iso:
9270
self._sqlserver_isolation_level = ISOLATION_LEVEL_MAP.get(effective_iso)
71+
else:
72+
self._sqlserver_isolation_level = None
9373

94-
await self._do_begin()
74+
# Base begin() increments _transaction_level first (so is_active is
75+
# already True while _do_begin() executes, preventing auto-commit),
76+
# issues a real BEGIN at level 1 and a savepoint for nested levels.
77+
await super().begin()
9578

96-
self._state = TransactionState.ACTIVE
97-
self._savepoints = []
79+
if self._transaction_level == 1:
80+
self._savepoints = []
9881

99-
self._log(logging.DEBUG, f"Async transaction started (isolation={self._sqlserver_isolation_level})")
82+
self.log(logging.DEBUG, f"Async transaction started (isolation={self._sqlserver_isolation_level})")
10083

10184
async def _do_begin(self) -> None:
102-
"""Execute the BEGIN TRANSACTION statement."""
85+
"""Execute the SET TRANSACTION ISOLATION LEVEL (if any) and BEGIN TRANSACTION."""
86+
if self._sqlserver_isolation_level:
87+
set_iso_sql = f"SET TRANSACTION ISOLATION LEVEL {self._sqlserver_isolation_level.value}"
88+
await self._backend.execute(set_iso_sql)
89+
10390
parts = ["BEGIN TRANSACTION"]
10491

10592
if self._name:
@@ -108,56 +95,68 @@ async def _do_begin(self) -> None:
10895
sql = " ".join(parts)
10996
await self._backend.execute(sql)
11097

111-
if self._sqlserver_isolation_level:
112-
set_iso_sql = f"SET TRANSACTION ISOLATION LEVEL {self._sqlserver_isolation_level.value}"
113-
await self._backend.execute(set_iso_sql)
98+
async def _do_commit(self) -> None:
99+
"""Execute the COMMIT TRANSACTION statement."""
100+
sql = "COMMIT TRANSACTION"
101+
if self._name:
102+
sql += f" {self._name}"
103+
await self._backend.execute(sql)
104+
105+
async def _do_rollback(self) -> None:
106+
"""Execute the ROLLBACK TRANSACTION statement."""
107+
sql = "ROLLBACK TRANSACTION"
108+
if self._name:
109+
sql += f" {self._name}"
110+
await self._backend.execute(sql)
111+
112+
async def _do_create_savepoint(self, name: str) -> None:
113+
"""Create a savepoint using SQL Server's SAVE TRANSACTION syntax."""
114+
quoted = self._backend.dialect.format_identifier(name)
115+
sql = f"SAVE TRANSACTION {quoted}"
116+
await self._backend.execute(sql)
117+
118+
async def _do_release_savepoint(self, name: str) -> None:
119+
"""Release a savepoint (no-op in SQL Server)."""
120+
if name in self._savepoints:
121+
self._savepoints.remove(name)
122+
123+
async def _do_rollback_savepoint(self, name: str) -> None:
124+
"""Rollback to a savepoint using SQL Server's ROLLBACK TRANSACTION syntax."""
125+
quoted = self._backend.dialect.format_identifier(name)
126+
sql = f"ROLLBACK TRANSACTION {quoted}"
127+
await self._backend.execute(sql)
114128

115129
async def commit(self) -> None:
116130
"""Commit the current async transaction."""
117-
if self._state != TransactionState.ACTIVE:
131+
if not self.is_active:
118132
raise TransactionError("No active transaction to commit")
119133

120-
self._log(logging.DEBUG, "Committing async transaction")
134+
self.log(logging.DEBUG, "Committing async transaction")
121135

122136
try:
123-
await self._do_commit()
124-
self._state = TransactionState.COMMITTED
125-
self._savepoints = []
126-
137+
await super().commit()
127138
except Exception as e:
128-
self._state = TransactionState.FAILED
129-
self._log(logging.ERROR, f"Failed to commit async transaction: {str(e)}")
139+
self.log(logging.ERROR, f"Failed to commit async transaction: {str(e)}")
130140
raise
131141

132-
async def _do_commit(self) -> None:
133-
"""Execute the COMMIT TRANSACTION statement."""
134-
sql = "COMMIT TRANSACTION"
135-
if self._name:
136-
sql += f" {self._name}"
137-
await self._backend.execute(sql)
138-
139142
async def rollback(self, savepoint: Optional[str] = None) -> None:
140143
"""Rollback the async transaction or to a savepoint.
141144
142145
Args:
143146
savepoint: If provided, rollback to this savepoint
144147
"""
145-
if self._state != TransactionState.ACTIVE:
148+
if not self.is_active:
146149
raise TransactionError("No active transaction to rollback")
147150

148151
if savepoint:
149152
await self._rollback_to_savepoint(savepoint)
150153
else:
151-
await self._do_rollback()
152-
self._state = TransactionState.ROLLED_BACK
153-
self._savepoints = []
154-
155-
async def _do_rollback(self) -> None:
156-
"""Execute the ROLLBACK TRANSACTION statement."""
157-
sql = "ROLLBACK TRANSACTION"
158-
if self._name:
159-
sql += f" {self._name}"
160-
await self._backend.execute(sql)
154+
self.log(logging.DEBUG, "Rolling back async transaction")
155+
try:
156+
await super().rollback()
157+
except Exception as e:
158+
self.log(logging.ERROR, f"Failed to rollback async transaction: {str(e)}")
159+
raise
161160

162161
async def savepoint(self, name: Optional[str] = None) -> str:
163162
"""Create a savepoint within the current async transaction.
@@ -175,27 +174,32 @@ async def savepoint(self, name: Optional[str] = None) -> str:
175174
name = f"sp_{len(self._savepoints) + 1}"
176175

177176
if name in self._savepoints:
178-
self._log(logging.WARNING, f"Savepoint '{name}' already exists")
177+
self.log(logging.WARNING, f"Savepoint '{name}' already exists")
179178

180-
sql = f"SAVE TRANSACTION {name}"
179+
quoted = self._backend.dialect.format_identifier(name)
180+
sql = f"SAVE TRANSACTION {quoted}"
181181
await self._backend.execute(sql)
182182
self._savepoints.append(name)
183+
self._active_savepoints.append(name)
183184

184-
self._log(logging.DEBUG, f"Async savepoint created: {name}")
185+
self.log(logging.DEBUG, f"Async savepoint created: {name}")
185186
return name
186187

187188
async def _rollback_to_savepoint(self, name: str) -> None:
188189
"""Rollback to a specific savepoint."""
189190
if name not in self._savepoints:
190191
raise TransactionError(f"Savepoint '{name}' not found")
191192

192-
sql = f"ROLLBACK TRANSACTION {name}"
193+
quoted = self._backend.dialect.format_identifier(name)
194+
sql = f"ROLLBACK TRANSACTION {quoted}"
193195
await self._backend.execute(sql)
194196

195197
idx = self._savepoints.index(name) + 1
196198
self._savepoints = self._savepoints[:idx]
199+
active_idx = self._active_savepoints.index(name) + 1 if name in self._active_savepoints else 0
200+
self._active_savepoints = self._active_savepoints[:active_idx]
197201

198-
self._log(logging.DEBUG, f"Rolled back to async savepoint: {name}")
202+
self.log(logging.DEBUG, f"Rolled back to async savepoint: {name}")
199203

200204
async def release_savepoint(self, name: str) -> None:
201205
"""Release a savepoint (no-op in SQL Server).
@@ -205,25 +209,10 @@ async def release_savepoint(self, name: str) -> None:
205209
"""
206210
if name in self._savepoints:
207211
self._savepoints.remove(name)
208-
self._log(logging.DEBUG, f"Async savepoint released: {name}")
212+
if name in self._active_savepoints:
213+
self._active_savepoints.remove(name)
214+
self.log(logging.DEBUG, f"Async savepoint released: {name}")
209215

210216
def savepoints(self) -> List[str]:
211217
"""Get list of current savepoints."""
212218
return self._savepoints.copy()
213-
214-
@asynccontextmanager
215-
async def transaction(self):
216-
"""Async context manager for transaction."""
217-
await self.begin()
218-
try:
219-
yield self
220-
except Exception:
221-
await self.rollback()
222-
raise
223-
else:
224-
await self.commit()
225-
226-
def _log(self, level: int, message: str) -> None:
227-
"""Log a message."""
228-
if self._logger:
229-
self._logger.log(level, message)

‎src/rhosocial/activerecord/backend/impl/sqlserver/backend.py‎

Lines changed: 26 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -285,19 +285,39 @@ def execute(
285285
DatabaseError: For other database errors
286286
"""
287287
from rhosocial.activerecord.backend.options import ExecutionOptions, StatementType
288-
288+
289+
sql_upper = sql.strip().upper()
290+
289291
if options is None:
290-
sql_upper = sql.strip().upper()
291292
if sql_upper.startswith(("SELECT", "WITH", "EXEC", "EXECUTE")):
292293
stmt_type = StatementType.DQL
293294
elif sql_upper.startswith(("INSERT", "UPDATE", "DELETE", "MERGE")):
294295
stmt_type = StatementType.DML
295296
else:
296297
stmt_type = StatementType.DDL
297-
298-
options = ExecutionOptions(stmt_type=stmt_type)
299-
300-
return super().execute(sql, params, options=options)
298+
299+
column_mapping = kwargs.get("column_mapping")
300+
column_adapters = kwargs.get("column_adapters")
301+
302+
options = ExecutionOptions(
303+
stmt_type=stmt_type,
304+
column_adapters=column_adapters,
305+
column_mapping=column_mapping,
306+
)
307+
else:
308+
if "column_mapping" in kwargs:
309+
options.column_mapping = kwargs["column_mapping"]
310+
if "column_adapters" in kwargs:
311+
options.column_adapters = kwargs["column_adapters"]
312+
313+
result = super().execute(sql, params, options=options)
314+
315+
if options.stmt_type == StatementType.DML and sql_upper.startswith("INSERT"):
316+
scope_id = self.get_scope_identity()
317+
if scope_id is not None:
318+
result.last_insert_id = scope_id
319+
320+
return result
301321

302322
def execute_many(self, sql: str, params_list: List[Tuple]) -> QueryResult:
303323
"""Execute the same SQL statement multiple times with different parameters.

‎src/rhosocial/activerecord/backend/impl/sqlserver/mixins/backend_mixin.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@ def set_identity_insert(self, table: str, on: bool = True) -> None:
2626

2727
def get_scope_identity(self) -> Optional[int]:
2828
cursor = self._get_cursor()
29-
cursor.execute("SELECT SCOPE_IDENTITY()")
29+
cursor.execute("SELECT @@IDENTITY")
3030
row = cursor.fetchone()
3131
cursor.close()
3232
if row and row[0] is not None:

0 commit comments

Comments
 (0)