|
6 | 6 | using aioodbc as the async driver wrapper. |
7 | 7 | """ |
8 | 8 |
|
9 | | -import asyncio |
10 | 9 | import logging |
11 | 10 | import time |
12 | 11 | from contextlib import asynccontextmanager |
13 | | -from typing import Any, AsyncIterator, Dict, List, Optional, Set, Tuple |
| 12 | +from typing import Any, AsyncIterator, List, Optional, Set, Tuple |
14 | 13 |
|
15 | 14 | from rhosocial.activerecord.backend.base import AsyncStorageBackend |
16 | 15 | from rhosocial.activerecord.backend.base.execution_logging import ( |
|
20 | 19 | log_execution_start, |
21 | 20 | log_execution_summary, |
22 | 21 | ) |
23 | | -from rhosocial.activerecord.backend.base.operations import AsyncSQLOperationsMixin |
24 | | -from rhosocial.activerecord.backend.base.transaction_management import AsyncTransactionManagementMixin |
25 | 22 | from rhosocial.activerecord.backend.errors import ( |
26 | 23 | ConnectionError, |
27 | 24 | DatabaseError, |
28 | | - DeadlockError, |
29 | | - IntegrityError, |
30 | | - OperationalError, |
31 | | - QueryError, |
| 25 | + TransactionError, |
32 | 26 | ) |
33 | 27 | from rhosocial.activerecord.backend.options import ExecutionOptions, StatementType |
34 | 28 | from rhosocial.activerecord.backend.result import QueryResult |
35 | 29 | from rhosocial.activerecord.logging.redaction import unwrap_sensitive |
36 | 30 | from rhosocial.activerecord.backend.introspection.backend_mixin import IntrospectorBackendMixin |
37 | 31 | from ..config import SQLServerConnectionConfig |
38 | 32 | from ..dialect import SQLServerDialect |
| 33 | +from ..error_mapping import classify_sqlserver_error |
39 | 34 | from ..async_transaction import AsyncSQLServerTransactionManager |
40 | 35 | from .backend import SQLServerUnicodeDialect |
41 | 36 | from ..mixins import SQLServerBackendMixin |
@@ -271,21 +266,59 @@ async def _get_cursor(self): |
271 | 266 |
|
272 | 267 | return await self._connection.cursor() |
273 | 268 |
|
| 269 | + async def _execute_control_sql_async(self, sql: str) -> None: |
| 270 | + """Run a session-control statement through the mapped error path.""" |
| 271 | + cursor = await self._get_cursor() |
| 272 | + try: |
| 273 | + await cursor.execute(sql) |
| 274 | + except Exception as exc: |
| 275 | + await self._handle_error(exc) |
| 276 | + finally: |
| 277 | + await self._close_cursor_quietly_async(cursor) |
| 278 | + |
| 279 | + async def _close_cursor_quietly_async(self, cursor) -> None: |
| 280 | + """Close an async cursor, logging rather than masking an in-flight error.""" |
| 281 | + try: |
| 282 | + await cursor.close() |
| 283 | + except Exception as exc: |
| 284 | + self.log(logging.DEBUG, f"Failed to close cursor: {exc}") |
| 285 | + |
| 286 | + async def _execute_control_sql_async_quietly(self, sql: str) -> None: |
| 287 | + """Run a cleanup statement without masking the original failure.""" |
| 288 | + try: |
| 289 | + await self._execute_control_sql_async(sql) |
| 290 | + except Exception as exc: |
| 291 | + self.log(logging.WARNING, f"Failed to reset session state ({sql}): {exc}") |
| 292 | + |
| 293 | + async def _reset_identity_insert_quietly(self, table: str) -> None: |
| 294 | + """Turn ``IDENTITY_INSERT`` off without replacing an in-flight failure.""" |
| 295 | + try: |
| 296 | + await self.set_identity_insert(table, False) |
| 297 | + except Exception as exc: |
| 298 | + self.log( |
| 299 | + logging.WARNING, |
| 300 | + f"Failed to disable IDENTITY_INSERT for {table}: {exc}", |
| 301 | + ) |
| 302 | + |
274 | 303 | async def _get_scope_identity_async(self) -> Optional[int]: |
275 | 304 | """Get the identity value of the last INSERT via @@IDENTITY.""" |
276 | 305 | sql, params = session_identity_sql(self.dialect) |
277 | | - async with await self._get_cursor() as cursor: |
| 306 | + cursor = await self._get_cursor() |
| 307 | + try: |
278 | 308 | await cursor.execute(sql, params) |
279 | 309 | row = await cursor.fetchone() |
| 310 | + except Exception as exc: |
| 311 | + await self._handle_error(exc) |
| 312 | + finally: |
| 313 | + await self._close_cursor_quietly_async(cursor) |
280 | 314 | if row and row[0] is not None: |
281 | 315 | return int(row[0]) |
282 | 316 | return None |
283 | 317 |
|
284 | 318 | async def set_identity_insert(self, table: str, on: bool = True) -> None: |
285 | 319 | """Enable or disable IDENTITY_INSERT for a table (async).""" |
286 | 320 | sql, _params = self.dialect.format_set_identity_insert(table, on) |
287 | | - async with await self._get_cursor() as cursor: |
288 | | - await cursor.execute(sql) |
| 321 | + await self._execute_control_sql_async(sql) |
289 | 322 |
|
290 | 323 | async def _get_identity_columns_async(self, table_name: str) -> Set[str]: |
291 | 324 | """Return the set of identity column names for a table (async, cached).""" |
@@ -352,13 +385,11 @@ async def execute( |
352 | 385 | explain_split = self._split_explain_statement(sql) |
353 | 386 | if explain_split is not None: |
354 | 387 | set_clause, statement = explain_split |
355 | | - async with await self._get_cursor() as cursor: |
356 | | - await cursor.execute(f"SET {set_clause} ON") |
| 388 | + await self._execute_control_sql_async(f"SET {set_clause} ON") |
357 | 389 | try: |
358 | 390 | return await self._execute_statement(statement, params, options) |
359 | 391 | finally: |
360 | | - async with await self._get_cursor() as cursor: |
361 | | - await cursor.execute(f"SET {set_clause} OFF") |
| 392 | + await self._execute_control_sql_async_quietly(f"SET {set_clause} OFF") |
362 | 393 |
|
363 | 394 | return await self._execute_statement(sql, params, options) |
364 | 395 |
|
@@ -419,9 +450,13 @@ async def _execute_statement( |
419 | 450 | await cursor.execute(final_sql, final_params or ()) |
420 | 451 | finally: |
421 | 452 | if identity_table is not None: |
422 | | - await self.set_identity_insert(identity_table, False) |
| 453 | + await self._reset_identity_insert_quietly(identity_table) |
423 | 454 |
|
424 | | - is_select = options.process_result_set if options.process_result_set is not None else options.stmt_type == StatementType.DQL |
| 455 | + is_select = ( |
| 456 | + options.process_result_set |
| 457 | + if options.process_result_set is not None |
| 458 | + else options.stmt_type == StatementType.DQL |
| 459 | + ) |
425 | 460 | data = None |
426 | 461 | if is_select and cursor.description: |
427 | 462 | rows = await cursor.fetchall() |
@@ -469,7 +504,10 @@ async def _execute_statement( |
469 | 504 |
|
470 | 505 | except Exception as e: |
471 | 506 | log_execution_error(self, e, options) |
| 507 | + if isinstance(e, DatabaseError): |
| 508 | + raise |
472 | 509 | await self._handle_error(e) |
| 510 | + raise DatabaseError(str(e)) from e |
473 | 511 |
|
474 | 512 | async def execute_many( |
475 | 513 | self, |
@@ -517,7 +555,11 @@ async def execute_many( |
517 | 555 |
|
518 | 556 | except Exception as e: |
519 | 557 | log_execution_error(self, e, options) |
| 558 | + if isinstance(e, DatabaseError): |
| 559 | + raise |
520 | 560 | await self._handle_error(e) |
| 561 | + # Same defensive fallback as the sync batch path. |
| 562 | + raise DatabaseError(str(e)) from e |
521 | 563 |
|
522 | 564 | async def get_server_version(self) -> Tuple[int, int, int]: |
523 | 565 | """Get SQL Server server version. |
@@ -617,62 +659,60 @@ async def executescript(self, sql_script: str) -> None: |
617 | 659 |
|
618 | 660 | for batch in batches: |
619 | 661 | batch = batch.strip() |
620 | | - if batch: |
621 | | - async with await self._get_cursor() as cursor: |
622 | | - await cursor.execute(batch) |
| 662 | + if not batch: |
| 663 | + continue |
| 664 | + async with await self._get_cursor() as cursor: |
| 665 | + await cursor.execute(batch) |
623 | 666 |
|
624 | 667 | await self._handle_auto_commit_if_needed() |
625 | 668 | duration = time.perf_counter() - start_time |
626 | 669 | self.log(logging.INFO, f"SQL script executed successfully, duration={duration:.3f}s") |
627 | 670 |
|
628 | 671 | except Exception as e: |
629 | 672 | self.log(logging.DEBUG, f"Error executing SQL script: {str(e)}") |
| 673 | + if isinstance(e, DatabaseError): |
| 674 | + raise |
630 | 675 | await self._handle_error(e) |
| 676 | + raise DatabaseError(str(e)) from e |
631 | 677 |
|
632 | 678 | async def _handle_error(self, error: Exception) -> None: |
633 | | - """Handle SQL Server-specific errors. |
634 | | - |
| 679 | + """Handle SQL Server-specific errors asynchronously. |
| 680 | +
|
| 681 | + Uses the same code -> class -> message recognition order as the sync |
| 682 | + backend, so both halves translate a driver failure identically. |
| 683 | +
|
635 | 684 | Args: |
636 | 685 | error: The exception to handle |
637 | | - |
| 686 | +
|
638 | 687 | Raises: |
639 | 688 | Appropriate backend exception |
640 | 689 | """ |
641 | | - error_msg = str(error).lower() |
642 | | - |
643 | | - if ( |
644 | | - 'integrity' in error_msg |
645 | | - or 'constraint' in error_msg |
646 | | - or 'does not allow nulls' in error_msg |
647 | | - or 'cannot insert the value null' in error_msg |
648 | | - ): |
649 | | - if "duplicate" in error_msg or "unique" in error_msg: |
650 | | - self.log(logging.DEBUG, f"Unique constraint violation: {str(error)}") |
651 | | - raise IntegrityError(f"Unique constraint violation: {str(error)}") |
652 | | - elif "foreign key" in error_msg or "reference" in error_msg: |
653 | | - self.log(logging.DEBUG, f"Foreign key constraint violation: {str(error)}") |
654 | | - raise IntegrityError(f"Foreign key constraint violation: {str(error)}") |
655 | | - self.log(logging.DEBUG, f"Integrity error: {str(error)}") |
656 | | - raise IntegrityError(self._normalize_integrity_message(str(error))) |
657 | | - |
658 | | - if 'deadlock' in error_msg or '1205' in error_msg: |
659 | | - self.log(logging.DEBUG, f"Deadlock detected: {str(error)}") |
660 | | - raise DeadlockError(str(error)) |
661 | | - |
662 | | - if 'timeout' in error_msg or 'connection' in error_msg: |
663 | | - self.log(logging.DEBUG, f"Operational error: {str(error)}") |
664 | | - raise OperationalError(str(error)) |
665 | | - |
666 | | - self.log(logging.DEBUG, f"Database error: {str(error)}") |
667 | | - raise DatabaseError(str(error)) |
| 690 | + # A framework error already carries its phase semantics; re-mapping it |
| 691 | + # would replace the original type with a generic driver error. |
| 692 | + if isinstance(error, DatabaseError): |
| 693 | + raise error |
| 694 | + mapped = classify_sqlserver_error(error) |
| 695 | + self._log_classified_error(error, mapped) |
| 696 | + raise mapped from error |
668 | 697 |
|
669 | 698 | async def _handle_auto_commit_if_needed(self) -> None: |
670 | 699 | """Handle auto-commit for async operations.""" |
671 | 700 | if not self.in_transaction and self._connection: |
672 | 701 | if not self.config.autocommit: |
673 | | - async with await self._get_cursor() as cursor: |
674 | | - await cursor.execute("COMMIT") |
675 | | - self.log(logging.DEBUG, "Auto-committed operation") |
| 702 | + await self._handle_auto_commit() |
| 703 | + |
| 704 | + async def _handle_auto_commit(self) -> None: |
| 705 | + """Commit outside a transaction, surfacing a failure as a transaction error.""" |
| 706 | + if self._connection is None or self.config.autocommit: |
| 707 | + return |
| 708 | + cursor = await self._get_cursor() |
| 709 | + try: |
| 710 | + await cursor.execute("COMMIT") |
| 711 | + except Exception as exc: |
| 712 | + raise TransactionError("Auto-commit failed") from exc |
| 713 | + finally: |
| 714 | + await self._close_cursor_quietly_async(cursor) |
| 715 | + self.log(logging.DEBUG, "Auto-committed operation") |
676 | 716 |
|
677 | 717 | def _create_introspector(self): |
678 | 718 | """Create a SQL Server introspector.""" |
|
0 commit comments