(self, connection: SqliteClient)
| 214 | |
| 215 | class SqliteTransactionWrapper(SqliteClient, TransactionalDBClient): |
| 216 | def __init__(self, connection: SqliteClient) -> None: |
| 217 | self.capabilities = connection.capabilities |
| 218 | self.connection_name = connection.connection_name |
| 219 | self._connection: aiosqlite.Connection = cast(aiosqlite.Connection, connection._connection) |
| 220 | self._lock = asyncio.Lock() |
| 221 | self._savepoint: str | None = None |
| 222 | self.log = connection.log |
| 223 | self._finalized = False |
| 224 | self.fetch_inserted = connection.fetch_inserted |
| 225 | self._parent = connection |
| 226 | |
| 227 | def _in_transaction(self) -> TransactionContext: |
| 228 | return NestedTransactionContext(SqliteTransactionWrapper(self)) |
nothing calls this directly
no outgoing calls
no test coverage detected