Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/sqlite/aiosqlite.py: 46%

1# dialects/sqlite/aiosqlite.py 

2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# 

5# This module is part of SQLAlchemy and is released under 

6# the MIT License: https://www.opensource.org/licenses/mit-license.php 

7 

8 

9r""" 

10 

11.. dialect:: sqlite+aiosqlite 

12 :name: aiosqlite 

13 :dbapi: aiosqlite 

14 :connectstring: sqlite+aiosqlite:///file_path 

15 :url: https://pypi.org/project/aiosqlite/ 

16 

17The aiosqlite dialect provides support for the SQLAlchemy asyncio interface 

18running on top of pysqlite. 

19 

20aiosqlite is a wrapper around pysqlite that uses a background thread for 

21each connection. It does not actually use non-blocking IO, as SQLite 

22databases are not socket-based. However it does provide a working asyncio 

23interface that's useful for testing and prototyping purposes. 

24 

25Using a special asyncio mediation layer, the aiosqlite dialect is usable 

26as the backend for the :ref:`SQLAlchemy asyncio <asyncio_toplevel>` 

27extension package. 

28 

29This dialect should normally be used only with the 

30:func:`_asyncio.create_async_engine` engine creation function:: 

31 

32 from sqlalchemy.ext.asyncio import create_async_engine 

33 

34 engine = create_async_engine("sqlite+aiosqlite:///filename") 

35 

36The URL passes through all arguments to the ``pysqlite`` driver, so all 

37connection arguments are the same as they are for that of :ref:`pysqlite`. 

38 

39.. _aiosqlite_udfs: 

40 

41User-Defined Functions 

42---------------------- 

43 

44aiosqlite extends pysqlite to support async, so we can create our own user-defined functions (UDFs) 

45in Python and use them directly in SQLite queries as described here: :ref:`pysqlite_udfs`. 

46 

47.. _aiosqlite_serializable: 

48 

49Serializable isolation / Savepoints / Transactional DDL (asyncio version) 

50------------------------------------------------------------------------- 

51 

52A newly revised version of this important section is now available 

53at the top level of the SQLAlchemy SQLite documentation, in the section 

54:ref:`sqlite_transactions`. 

55 

56 

57.. _aiosqlite_pooling: 

58 

59Pooling Behavior 

60---------------- 

61 

62The SQLAlchemy ``aiosqlite`` DBAPI establishes the connection pool differently 

63based on the kind of SQLite database that's requested: 

64 

65* When a ``:memory:`` SQLite database is specified, the dialect by default 

66 will use :class:`.StaticPool`. This pool maintains a single 

67 connection, so that all access to the engine 

68 use the same ``:memory:`` database. 

69* When a file-based database is specified, the dialect will use 

70 :class:`.AsyncAdaptedQueuePool` as the source of connections. 

71 

72 .. versionchanged:: 2.0.38 

73 

74 SQLite file database engines now use :class:`.AsyncAdaptedQueuePool` by default. 

75 Previously, :class:`.NullPool` were used. The :class:`.NullPool` class 

76 may be used by specifying it via the 

77 :paramref:`_sa.create_engine.poolclass` parameter. 

78 

79As with the pysqlite dialect, this selection is made based on the database 

80name alone, and the ``mode=memory`` query string argument is deprecated as 

81a means of influencing it; see :ref:`pysqlite_threading_pooling` for 

82background. 

83 

84.. _aiosqlite_memory: 

85 

86Using a Memory Database with Multiple Coroutines 

87------------------------------------------------- 

88 

89The default :class:`.StaticPool` used for ``:memory:`` databases forces all 

90coroutines to share a single DBAPI connection. Because SQLite maintains only 

91one transaction state per connection, concurrent coroutines can interfere 

92with each other — a ``ROLLBACK`` in one coroutine will also discard 

93uncommitted work from any other coroutine using the same engine. 

94 

95For async workloads where multiple :class:`.AsyncSession` or 

96:class:`.AsyncConnection` objects may be active simultaneously, use SQLite's 

97shared-cache URI mode instead. This gives each checkout its own DBAPI 

98connection with independent transaction state while still sharing one 

99in-memory database:: 

100 

101 engine = create_async_engine( 

102 "sqlite+aiosqlite:///file::memory:?cache=shared&uri=true" 

103 ) 

104 

105Because this URL form is treated as a file-based database by the dialect, 

106:class:`.AsyncAdaptedQueuePool` is used automatically and no additional 

107configuration is needed. 

108 

109Note that a shared-cache database is discarded once its last connection is 

110closed, so that operations such as :meth:`_asyncio.AsyncEngine.dispose` or 

111the use of :paramref:`_sa.create_engine.pool_recycle` will destroy its 

112contents; see :ref:`pysqlite_shared_cache_lifespan` for background and for 

113how to hold such a database open. 

114 

115See the pysqlite documentation at 

116:ref:`pysqlite_uri_shared_cache` for full details on shared-cache memory 

117databases, including how to use named databases to maintain multiple 

118independent in-memory databases within the same process. 

119 

120""" # noqa 

121 

122from __future__ import annotations 

123 

124import asyncio 

125from functools import partial 

126from threading import Thread 

127from types import ModuleType 

128from typing import Any 

129from typing import cast 

130from typing import NoReturn 

131from typing import Optional 

132from typing import TYPE_CHECKING 

133from typing import Union 

134 

135from .base import SQLiteExecutionContext 

136from .pysqlite import SQLiteDialect_pysqlite 

137from ... import pool 

138from ... import util 

139from ...connectors.asyncio import AsyncAdapt_dbapi_connection 

140from ...connectors.asyncio import AsyncAdapt_dbapi_cursor 

141from ...connectors.asyncio import AsyncAdapt_dbapi_module 

142from ...connectors.asyncio import AsyncAdapt_dbapi_ss_cursor 

143from ...connectors.asyncio import AsyncAdapt_terminate 

144from ...util.concurrency import await_ 

145 

146if TYPE_CHECKING: 

147 from ...connectors.asyncio import AsyncIODBAPIConnection 

148 from ...engine.interfaces import DBAPIConnection 

149 from ...engine.interfaces import DBAPICursor 

150 from ...engine.interfaces import DBAPIModule 

151 from ...engine.url import URL 

152 from ...pool.base import PoolProxiedConnection 

153 

154 

155class AsyncAdapt_aiosqlite_cursor(AsyncAdapt_dbapi_cursor): 

156 __slots__ = () 

157 

158 

159class AsyncAdapt_aiosqlite_ss_cursor(AsyncAdapt_dbapi_ss_cursor): 

160 __slots__ = () 

161 

162 

163class AsyncAdapt_aiosqlite_connection( 

164 AsyncAdapt_terminate, AsyncAdapt_dbapi_connection 

165): 

166 __slots__ = () 

167 

168 _cursor_cls = AsyncAdapt_aiosqlite_cursor 

169 _ss_cursor_cls = AsyncAdapt_aiosqlite_ss_cursor 

170 

171 @property 

172 def isolation_level(self) -> Optional[str]: 

173 return cast(str, self._connection.isolation_level) 

174 

175 @isolation_level.setter 

176 def isolation_level(self, value: Optional[str]) -> None: 

177 # aiosqlite's isolation_level setter works outside the Thread 

178 # that it's supposed to, necessitating setting check_same_thread=False. 

179 # for improved stability, we instead invent our own awaitable version 

180 # using aiosqlite's async queue directly. 

181 

182 def set_iso( 

183 connection: AsyncAdapt_aiosqlite_connection, value: Optional[str] 

184 ) -> None: 

185 connection.isolation_level = value 

186 

187 function = partial(set_iso, self._connection._conn, value) 

188 future = asyncio.get_event_loop().create_future() 

189 

190 self._connection._tx.put_nowait((future, function)) 

191 

192 try: 

193 await_(future) 

194 except Exception as error: 

195 self._handle_exception(error) 

196 

197 def create_function(self, *args: Any, **kw: Any) -> None: 

198 try: 

199 await_(self._connection.create_function(*args, **kw)) 

200 except Exception as error: 

201 self._handle_exception(error) 

202 

203 def rollback(self) -> None: 

204 if self._connection._connection: 

205 super().rollback() 

206 

207 def commit(self) -> None: 

208 if self._connection._connection: 

209 super().commit() 

210 

211 def close(self) -> None: 

212 try: 

213 await_(self._connection.close()) 

214 except ValueError: 

215 # this is undocumented for aiosqlite, that ValueError 

216 # was raised if .close() was called more than once, which is 

217 # both not customary for DBAPI and is also not a DBAPI.Error 

218 # exception. This is now fixed in aiosqlite via my PR 

219 # https://github.com/omnilib/aiosqlite/pull/238, so we can be 

220 # assured this will not become some other kind of exception, 

221 # since it doesn't raise anymore. 

222 

223 pass 

224 except Exception as error: 

225 self._handle_exception(error) 

226 

227 @classmethod 

228 def _handle_exception_no_connection( 

229 cls, dbapi: Any, error: Exception 

230 ) -> NoReturn: 

231 if isinstance(error, ValueError) and error.args[0].lower() in ( 

232 "no active connection", 

233 "connection closed", 

234 ): 

235 raise dbapi.sqlite.OperationalError(error.args[0]) from error 

236 else: 

237 super()._handle_exception_no_connection(dbapi, error) 

238 

239 async def _terminate_graceful_close(self) -> None: 

240 """Try to close connection gracefully""" 

241 await self._connection.close() 

242 

243 def _terminate_force_close(self) -> None: 

244 """Terminate the connection""" 

245 

246 # this was added in aiosqlite 0.22.1. if stop() is not present, 

247 # the dialect should indicate has_terminate=False 

248 try: 

249 meth = self._connection.stop 

250 except AttributeError as ae: 

251 raise NotImplementedError( 

252 "terminate_force_close() not implemented by this DBAPI shim" 

253 ) from ae 

254 else: 

255 meth() 

256 

257 

258class AsyncAdapt_aiosqlite_dbapi(AsyncAdapt_dbapi_module): 

259 def __init__(self, aiosqlite: ModuleType, sqlite: ModuleType): 

260 super().__init__(aiosqlite, dbapi_module=sqlite) 

261 self.aiosqlite = aiosqlite 

262 self.sqlite = sqlite 

263 self.paramstyle = "qmark" 

264 self.has_stop = hasattr(aiosqlite.Connection, "stop") 

265 self._init_dbapi_attributes() 

266 

267 def _init_dbapi_attributes(self) -> None: 

268 for name in ( 

269 "DatabaseError", 

270 "Error", 

271 "IntegrityError", 

272 "NotSupportedError", 

273 "OperationalError", 

274 "ProgrammingError", 

275 "sqlite_version", 

276 "sqlite_version_info", 

277 ): 

278 setattr(self, name, getattr(self.aiosqlite, name)) 

279 

280 for name in ("PARSE_COLNAMES", "PARSE_DECLTYPES"): 

281 setattr(self, name, getattr(self.sqlite, name)) 

282 

283 for name in ("Binary",): 

284 setattr(self, name, getattr(self.sqlite, name)) 

285 

286 def connect(self, *arg: Any, **kw: Any) -> AsyncAdapt_aiosqlite_connection: 

287 creator_fn = kw.pop("async_creator_fn", None) 

288 if creator_fn: 

289 connection = creator_fn(*arg, **kw) 

290 else: 

291 connection = self.aiosqlite.connect(*arg, **kw) 

292 

293 # aiosqlite uses a Thread. you'll thank us later 

294 if isinstance(connection, Thread): 

295 # Connection itself was a thread in version prior to 0.22 

296 connection.daemon = True 

297 else: 

298 # in 0.22+ instead it contains a thread. 

299 connection._thread.daemon = True 

300 

301 return AsyncAdapt_aiosqlite_connection(self, await_(connection)) 

302 

303 

304class SQLiteExecutionContext_aiosqlite(SQLiteExecutionContext): 

305 def create_server_side_cursor(self) -> DBAPICursor: 

306 return self._dbapi_connection.cursor(server_side=True) 

307 

308 

309class SQLiteDialect_aiosqlite(SQLiteDialect_pysqlite): 

310 driver = "aiosqlite" 

311 supports_statement_cache = True 

312 

313 is_async = True 

314 has_terminate = True 

315 

316 supports_server_side_cursors = True 

317 

318 execution_ctx_cls = SQLiteExecutionContext_aiosqlite 

319 

320 def __init__(self, **kwargs: Any): 

321 super().__init__(**kwargs) 

322 if self.dbapi and not self.dbapi.has_stop: 

323 self.has_terminate = False 

324 

325 @classmethod 

326 def import_dbapi(cls) -> AsyncAdapt_aiosqlite_dbapi: 

327 return AsyncAdapt_aiosqlite_dbapi( 

328 __import__("aiosqlite"), __import__("sqlite3") 

329 ) 

330 

331 def retrieve_dbapi_version(self, dbapi: DBAPIModule) -> util.VersionInfo: 

332 # the version of aiosqlite, rather than the Python version 

333 # reported by the pysqlite dialect 

334 aiosqlite = getattr(dbapi, "aiosqlite", None) 

335 return util.parse_version_string( 

336 getattr(aiosqlite, "__version__", None) 

337 ) 

338 

339 @classmethod 

340 def get_pool_class(cls, url: URL) -> type[pool.Pool]: 

341 if cls._is_url_file_db(url): 

342 return pool.AsyncAdaptedQueuePool 

343 else: 

344 cls._warn_memory_mode_pool_selection( 

345 url, pool.StaticPool, pool.AsyncAdaptedQueuePool 

346 ) 

347 return pool.StaticPool 

348 

349 def is_disconnect( 

350 self, 

351 e: DBAPIModule.Error, 

352 connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]], 

353 cursor: Optional[DBAPICursor], 

354 ) -> bool: 

355 self.dbapi = cast("DBAPIModule", self.dbapi) 

356 if isinstance(e, self.dbapi.OperationalError): 

357 err_lower = str(e).lower() 

358 if ( 

359 "no active connection" in err_lower 

360 or "connection closed" in err_lower 

361 ): 

362 return True 

363 

364 return super().is_disconnect(e, connection, cursor) 

365 

366 def get_driver_connection( 

367 self, connection: DBAPIConnection 

368 ) -> AsyncIODBAPIConnection: 

369 return connection._connection # type: ignore[no-any-return] 

370 

371 def do_terminate(self, dbapi_connection: DBAPIConnection) -> None: 

372 dbapi_connection.terminate() 

373 

374 

375dialect = SQLiteDialect_aiosqlite