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 

79.. _aiosqlite_memory: 

80 

81Using a Memory Database with Multiple Coroutines 

82------------------------------------------------- 

83 

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

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

86one transaction state per connection, concurrent coroutines can interfere 

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

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

89 

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

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

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

93connection with independent transaction state while still sharing one 

94in-memory database:: 

95 

96 engine = create_async_engine( 

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

98 ) 

99 

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

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

102configuration is needed. 

103 

104See the pysqlite documentation at 

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

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

107independent in-memory databases within the same process. 

108 

109""" # noqa 

110 

111from __future__ import annotations 

112 

113import asyncio 

114from functools import partial 

115from threading import Thread 

116from types import ModuleType 

117from typing import Any 

118from typing import cast 

119from typing import NoReturn 

120from typing import Optional 

121from typing import TYPE_CHECKING 

122from typing import Union 

123 

124from .base import SQLiteExecutionContext 

125from .pysqlite import SQLiteDialect_pysqlite 

126from ... import pool 

127from ... import util 

128from ...connectors.asyncio import AsyncAdapt_dbapi_connection 

129from ...connectors.asyncio import AsyncAdapt_dbapi_cursor 

130from ...connectors.asyncio import AsyncAdapt_dbapi_module 

131from ...connectors.asyncio import AsyncAdapt_dbapi_ss_cursor 

132from ...connectors.asyncio import AsyncAdapt_terminate 

133from ...util.concurrency import await_ 

134 

135if TYPE_CHECKING: 

136 from ...connectors.asyncio import AsyncIODBAPIConnection 

137 from ...engine.interfaces import DBAPIConnection 

138 from ...engine.interfaces import DBAPICursor 

139 from ...engine.interfaces import DBAPIModule 

140 from ...engine.url import URL 

141 from ...pool.base import PoolProxiedConnection 

142 

143 

144class AsyncAdapt_aiosqlite_cursor(AsyncAdapt_dbapi_cursor): 

145 __slots__ = () 

146 

147 

148class AsyncAdapt_aiosqlite_ss_cursor(AsyncAdapt_dbapi_ss_cursor): 

149 __slots__ = () 

150 

151 

152class AsyncAdapt_aiosqlite_connection( 

153 AsyncAdapt_terminate, AsyncAdapt_dbapi_connection 

154): 

155 __slots__ = () 

156 

157 _cursor_cls = AsyncAdapt_aiosqlite_cursor 

158 _ss_cursor_cls = AsyncAdapt_aiosqlite_ss_cursor 

159 

160 @property 

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

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

163 

164 @isolation_level.setter 

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

166 # aiosqlite's isolation_level setter works outside the Thread 

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

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

169 # using aiosqlite's async queue directly. 

170 

171 def set_iso( 

172 connection: AsyncAdapt_aiosqlite_connection, value: Optional[str] 

173 ) -> None: 

174 connection.isolation_level = value 

175 

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

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

178 

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

180 

181 try: 

182 await_(future) 

183 except Exception as error: 

184 self._handle_exception(error) 

185 

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

187 try: 

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

189 except Exception as error: 

190 self._handle_exception(error) 

191 

192 def rollback(self) -> None: 

193 if self._connection._connection: 

194 super().rollback() 

195 

196 def commit(self) -> None: 

197 if self._connection._connection: 

198 super().commit() 

199 

200 def close(self) -> None: 

201 try: 

202 await_(self._connection.close()) 

203 except ValueError: 

204 # this is undocumented for aiosqlite, that ValueError 

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

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

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

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

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

210 # since it doesn't raise anymore. 

211 

212 pass 

213 except Exception as error: 

214 self._handle_exception(error) 

215 

216 @classmethod 

217 def _handle_exception_no_connection( 

218 cls, dbapi: Any, error: Exception 

219 ) -> NoReturn: 

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

221 "no active connection", 

222 "connection closed", 

223 ): 

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

225 else: 

226 super()._handle_exception_no_connection(dbapi, error) 

227 

228 async def _terminate_graceful_close(self) -> None: 

229 """Try to close connection gracefully""" 

230 await self._connection.close() 

231 

232 def _terminate_force_close(self) -> None: 

233 """Terminate the connection""" 

234 

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

236 # the dialect should indicate has_terminate=False 

237 try: 

238 meth = self._connection.stop 

239 except AttributeError as ae: 

240 raise NotImplementedError( 

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

242 ) from ae 

243 else: 

244 meth() 

245 

246 

247class AsyncAdapt_aiosqlite_dbapi(AsyncAdapt_dbapi_module): 

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

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

250 self.aiosqlite = aiosqlite 

251 self.sqlite = sqlite 

252 self.paramstyle = "qmark" 

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

254 self._init_dbapi_attributes() 

255 

256 def _init_dbapi_attributes(self) -> None: 

257 for name in ( 

258 "DatabaseError", 

259 "Error", 

260 "IntegrityError", 

261 "NotSupportedError", 

262 "OperationalError", 

263 "ProgrammingError", 

264 "sqlite_version", 

265 "sqlite_version_info", 

266 ): 

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

268 

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

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

271 

272 for name in ("Binary",): 

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

274 

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

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

277 if creator_fn: 

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

279 else: 

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

281 

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

283 if isinstance(connection, Thread): 

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

285 connection.daemon = True 

286 else: 

287 # in 0.22+ instead it contains a thread. 

288 connection._thread.daemon = True 

289 

290 return AsyncAdapt_aiosqlite_connection(self, await_(connection)) 

291 

292 

293class SQLiteExecutionContext_aiosqlite(SQLiteExecutionContext): 

294 def create_server_side_cursor(self) -> DBAPICursor: 

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

296 

297 

298class SQLiteDialect_aiosqlite(SQLiteDialect_pysqlite): 

299 driver = "aiosqlite" 

300 supports_statement_cache = True 

301 

302 is_async = True 

303 has_terminate = True 

304 

305 supports_server_side_cursors = True 

306 

307 execution_ctx_cls = SQLiteExecutionContext_aiosqlite 

308 

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

310 super().__init__(**kwargs) 

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

312 self.has_terminate = False 

313 

314 @classmethod 

315 def import_dbapi(cls) -> AsyncAdapt_aiosqlite_dbapi: 

316 return AsyncAdapt_aiosqlite_dbapi( 

317 __import__("aiosqlite"), __import__("sqlite3") 

318 ) 

319 

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

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

322 # reported by the pysqlite dialect 

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

324 return util.parse_version_string( 

325 getattr(aiosqlite, "__version__", None) 

326 ) 

327 

328 @classmethod 

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

330 if cls._is_url_file_db(url): 

331 return pool.AsyncAdaptedQueuePool 

332 else: 

333 return pool.StaticPool 

334 

335 def is_disconnect( 

336 self, 

337 e: DBAPIModule.Error, 

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

339 cursor: Optional[DBAPICursor], 

340 ) -> bool: 

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

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

343 err_lower = str(e).lower() 

344 if ( 

345 "no active connection" in err_lower 

346 or "connection closed" in err_lower 

347 ): 

348 return True 

349 

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

351 

352 def get_driver_connection( 

353 self, connection: DBAPIConnection 

354 ) -> AsyncIODBAPIConnection: 

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

356 

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

358 dbapi_connection.terminate() 

359 

360 

361dialect = SQLiteDialect_aiosqlite