Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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
9r"""
11.. dialect:: sqlite+aiosqlite
12 :name: aiosqlite
13 :dbapi: aiosqlite
14 :connectstring: sqlite+aiosqlite:///file_path
15 :url: https://pypi.org/project/aiosqlite/
17The aiosqlite dialect provides support for the SQLAlchemy asyncio interface
18running on top of pysqlite.
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.
25Using a special asyncio mediation layer, the aiosqlite dialect is usable
26as the backend for the :ref:`SQLAlchemy asyncio <asyncio_toplevel>`
27extension package.
29This dialect should normally be used only with the
30:func:`_asyncio.create_async_engine` engine creation function::
32 from sqlalchemy.ext.asyncio import create_async_engine
34 engine = create_async_engine("sqlite+aiosqlite:///filename")
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`.
39.. _aiosqlite_udfs:
41User-Defined Functions
42----------------------
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`.
47.. _aiosqlite_serializable:
49Serializable isolation / Savepoints / Transactional DDL (asyncio version)
50-------------------------------------------------------------------------
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`.
57.. _aiosqlite_pooling:
59Pooling Behavior
60----------------
62The SQLAlchemy ``aiosqlite`` DBAPI establishes the connection pool differently
63based on the kind of SQLite database that's requested:
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.
72 .. versionchanged:: 2.0.38
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.
79.. _aiosqlite_memory:
81Using a Memory Database with Multiple Coroutines
82-------------------------------------------------
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.
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::
96 engine = create_async_engine(
97 "sqlite+aiosqlite:///file::memory:?cache=shared&uri=true"
98 )
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.
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.
109""" # noqa
111from __future__ import annotations
113import asyncio
114from collections import deque
115from functools import partial
116from threading import Thread
117from types import ModuleType
118from typing import Any
119from typing import cast
120from typing import Deque
121from typing import Iterator
122from typing import NoReturn
123from typing import Optional
124from typing import Sequence
125from typing import TYPE_CHECKING
126from typing import Union
128from .base import SQLiteExecutionContext
129from .pysqlite import SQLiteDialect_pysqlite
130from ... import pool
131from ... import util
132from ...connectors.asyncio import AsyncAdapt_dbapi_module
133from ...connectors.asyncio import AsyncAdapt_terminate
134from ...engine import AdaptedConnection
135from ...util.concurrency import await_fallback
136from ...util.concurrency import await_only
138if TYPE_CHECKING:
139 from ...connectors.asyncio import AsyncIODBAPIConnection
140 from ...connectors.asyncio import AsyncIODBAPICursor
141 from ...engine.interfaces import _DBAPICursorDescription
142 from ...engine.interfaces import _DBAPIMultiExecuteParams
143 from ...engine.interfaces import _DBAPISingleExecuteParams
144 from ...engine.interfaces import DBAPIConnection
145 from ...engine.interfaces import DBAPICursor
146 from ...engine.interfaces import DBAPIModule
147 from ...engine.url import URL
148 from ...pool.base import PoolProxiedConnection
151class AsyncAdapt_aiosqlite_cursor:
152 # TODO: base on connectors/asyncio.py
153 # see #10415
155 __slots__ = (
156 "_adapt_connection",
157 "_connection",
158 "description",
159 "await_",
160 "_rows",
161 "arraysize",
162 "rowcount",
163 "lastrowid",
164 )
166 server_side = False
168 def __init__(self, adapt_connection: AsyncAdapt_aiosqlite_connection):
169 self._adapt_connection = adapt_connection
170 self._connection = adapt_connection._connection
171 self.await_ = adapt_connection.await_
172 self.arraysize = 1
173 self.rowcount = -1
174 self.description: Optional[_DBAPICursorDescription] = None
175 self._rows: Deque[Any] = deque()
177 async def _async_soft_close(self) -> None:
178 return
180 def close(self) -> None:
181 self._rows.clear()
183 def execute(
184 self,
185 operation: Any,
186 parameters: Optional[_DBAPISingleExecuteParams] = None,
187 ) -> Any:
189 try:
190 _cursor: AsyncIODBAPICursor = self.await_(self._connection.cursor()) # type: ignore[arg-type] # noqa: E501
192 if parameters is None:
193 self.await_(_cursor.execute(operation))
194 else:
195 self.await_(_cursor.execute(operation, parameters))
197 if _cursor.description:
198 self.description = _cursor.description
199 self.lastrowid = self.rowcount = -1
201 if not self.server_side:
202 self._rows = deque(self.await_(_cursor.fetchall()))
203 else:
204 self.description = None
205 self.lastrowid = _cursor.lastrowid
206 self.rowcount = _cursor.rowcount
208 if not self.server_side:
209 self.await_(_cursor.close())
210 else:
211 self._cursor = _cursor # type: ignore[misc]
212 except Exception as error:
213 self._adapt_connection._handle_exception(error)
215 def executemany(
216 self,
217 operation: Any,
218 seq_of_parameters: _DBAPIMultiExecuteParams,
219 ) -> Any:
220 try:
221 _cursor: AsyncIODBAPICursor = self.await_(self._connection.cursor()) # type: ignore[arg-type] # noqa: E501
222 self.await_(_cursor.executemany(operation, seq_of_parameters))
223 self.description = None
224 self.lastrowid = _cursor.lastrowid
225 self.rowcount = _cursor.rowcount
226 self.await_(_cursor.close())
227 except Exception as error:
228 self._adapt_connection._handle_exception(error)
230 def setinputsizes(self, *inputsizes: Any) -> None:
231 pass
233 def __iter__(self) -> Iterator[Any]:
234 while self._rows:
235 yield self._rows.popleft()
237 def fetchone(self) -> Optional[Any]:
238 if self._rows:
239 return self._rows.popleft()
240 else:
241 return None
243 def fetchmany(self, size: Optional[int] = None) -> Sequence[Any]:
244 if size is None:
245 size = self.arraysize
247 rr = self._rows
248 return [rr.popleft() for _ in range(min(size, len(rr)))]
250 def fetchall(self) -> Sequence[Any]:
251 retval = list(self._rows)
252 self._rows.clear()
253 return retval
256class AsyncAdapt_aiosqlite_ss_cursor(AsyncAdapt_aiosqlite_cursor):
257 # TODO: base on connectors/asyncio.py
258 # see #10415
259 __slots__ = "_cursor"
261 server_side = True
263 def __init__(self, *arg: Any, **kw: Any) -> None:
264 super().__init__(*arg, **kw)
265 self._cursor: Optional[AsyncIODBAPICursor] = None
267 def close(self) -> None:
268 if self._cursor is not None:
269 self.await_(self._cursor.close())
270 self._cursor = None
272 def fetchone(self) -> Optional[Any]:
273 assert self._cursor is not None
274 return self.await_(self._cursor.fetchone())
276 def fetchmany(self, size: Optional[int] = None) -> Sequence[Any]:
277 assert self._cursor is not None
278 if size is None:
279 size = self.arraysize
280 return self.await_(self._cursor.fetchmany(size=size))
282 def fetchall(self) -> Sequence[Any]:
283 assert self._cursor is not None
284 return self.await_(self._cursor.fetchall())
287class AsyncAdapt_aiosqlite_connection(AsyncAdapt_terminate, AdaptedConnection):
288 await_ = staticmethod(await_only)
289 __slots__ = ("dbapi",)
291 def __init__(self, dbapi: Any, connection: AsyncIODBAPIConnection) -> None:
292 self.dbapi = dbapi
293 self._connection = connection
295 @property
296 def isolation_level(self) -> Optional[str]:
297 return cast(str, self._connection.isolation_level)
299 @isolation_level.setter
300 def isolation_level(self, value: Optional[str]) -> None:
301 # aiosqlite's isolation_level setter works outside the Thread
302 # that it's supposed to, necessitating setting check_same_thread=False.
303 # for improved stability, we instead invent our own awaitable version
304 # using aiosqlite's async queue directly.
306 def set_iso(
307 connection: AsyncAdapt_aiosqlite_connection, value: Optional[str]
308 ) -> None:
309 connection.isolation_level = value
311 function = partial(set_iso, self._connection._conn, value)
312 future = asyncio.get_event_loop().create_future()
314 self._connection._tx.put_nowait((future, function))
316 try:
317 self.await_(future)
318 except Exception as error:
319 self._handle_exception(error)
321 def create_function(self, *args: Any, **kw: Any) -> None:
322 try:
323 self.await_(self._connection.create_function(*args, **kw))
324 except Exception as error:
325 self._handle_exception(error)
327 def cursor(self, server_side: bool = False) -> AsyncAdapt_aiosqlite_cursor:
328 if server_side:
329 return AsyncAdapt_aiosqlite_ss_cursor(self)
330 else:
331 return AsyncAdapt_aiosqlite_cursor(self)
333 def execute(self, *args: Any, **kw: Any) -> Any:
334 return self.await_(self._connection.execute(*args, **kw))
336 def rollback(self) -> None:
337 try:
338 self.await_(self._connection.rollback())
339 except Exception as error:
340 self._handle_exception(error)
342 def commit(self) -> None:
343 try:
344 self.await_(self._connection.commit())
345 except Exception as error:
346 self._handle_exception(error)
348 def close(self) -> None:
349 try:
350 self.await_(self._connection.close())
351 except ValueError:
352 # this is undocumented for aiosqlite, that ValueError
353 # was raised if .close() was called more than once, which is
354 # both not customary for DBAPI and is also not a DBAPI.Error
355 # exception. This is now fixed in aiosqlite via my PR
356 # https://github.com/omnilib/aiosqlite/pull/238, so we can be
357 # assured this will not become some other kind of exception,
358 # since it doesn't raise anymore.
360 pass
361 except Exception as error:
362 self._handle_exception(error)
364 def _handle_exception(self, error: Exception) -> NoReturn:
365 if (
366 isinstance(error, ValueError)
367 and error.args[0] == "no active connection"
368 ):
369 raise self.dbapi.sqlite.OperationalError(
370 "no active connection"
371 ) from error
372 else:
373 raise error
375 async def _terminate_graceful_close(self) -> None:
376 """Try to close connection gracefully"""
377 await self._connection.close()
379 def _terminate_force_close(self) -> None:
380 """Terminate the connection"""
382 # this was added in aiosqlite 0.22.1. if stop() is not present,
383 # the dialect should indicate has_terminate=False
384 try:
385 meth = self._connection.stop
386 except AttributeError as ae:
387 raise NotImplementedError(
388 "terminate_force_close() not implemented by this DBAPI shim"
389 ) from ae
390 else:
391 meth()
394class AsyncAdaptFallback_aiosqlite_connection(AsyncAdapt_aiosqlite_connection):
395 __slots__ = ()
397 await_ = staticmethod(await_fallback)
400class AsyncAdapt_aiosqlite_dbapi(AsyncAdapt_dbapi_module):
401 def __init__(self, aiosqlite: ModuleType, sqlite: ModuleType):
402 self.aiosqlite = aiosqlite
403 self.sqlite = sqlite
404 self.paramstyle = "qmark"
405 self.has_stop = hasattr(aiosqlite.Connection, "stop")
406 self._init_dbapi_attributes()
408 def _init_dbapi_attributes(self) -> None:
409 for name in (
410 "DatabaseError",
411 "Error",
412 "IntegrityError",
413 "NotSupportedError",
414 "OperationalError",
415 "ProgrammingError",
416 "sqlite_version",
417 "sqlite_version_info",
418 ):
419 setattr(self, name, getattr(self.aiosqlite, name))
421 for name in ("PARSE_COLNAMES", "PARSE_DECLTYPES"):
422 setattr(self, name, getattr(self.sqlite, name))
424 for name in ("Binary",):
425 setattr(self, name, getattr(self.sqlite, name))
427 def connect(self, *arg: Any, **kw: Any) -> AsyncAdapt_aiosqlite_connection:
428 async_fallback = kw.pop("async_fallback", False)
430 creator_fn = kw.pop("async_creator_fn", None)
431 if creator_fn:
432 connection = creator_fn(*arg, **kw)
433 else:
434 connection = self.aiosqlite.connect(*arg, **kw)
436 # aiosqlite uses a Thread. you'll thank us later
437 if isinstance(connection, Thread):
438 # Connection itself was a thread in version prior to 0.22
439 connection.daemon = True
440 else:
441 # in 0.22+ instead it contains a thread.
442 connection._thread.daemon = True
444 if util.asbool(async_fallback):
445 return AsyncAdaptFallback_aiosqlite_connection(
446 self,
447 await_fallback(connection),
448 )
449 else:
450 return AsyncAdapt_aiosqlite_connection(
451 self,
452 await_only(connection),
453 )
456class SQLiteExecutionContext_aiosqlite(SQLiteExecutionContext):
457 def create_server_side_cursor(self) -> DBAPICursor:
458 return self._dbapi_connection.cursor(server_side=True)
461class SQLiteDialect_aiosqlite(SQLiteDialect_pysqlite):
462 driver = "aiosqlite"
463 supports_statement_cache = True
465 is_async = True
466 has_terminate = True
468 supports_server_side_cursors = True
470 execution_ctx_cls = SQLiteExecutionContext_aiosqlite
472 def __init__(self, **kwargs: Any):
473 super().__init__(**kwargs)
474 if self.dbapi and not self.dbapi.has_stop:
475 self.has_terminate = False
477 @classmethod
478 def import_dbapi(cls) -> AsyncAdapt_aiosqlite_dbapi:
479 return AsyncAdapt_aiosqlite_dbapi(
480 __import__("aiosqlite"), __import__("sqlite3")
481 )
483 @classmethod
484 def get_pool_class(cls, url: URL) -> type[pool.Pool]:
485 if cls._is_url_file_db(url):
486 return pool.AsyncAdaptedQueuePool
487 else:
488 return pool.StaticPool
490 def is_disconnect(
491 self,
492 e: DBAPIModule.Error,
493 connection: Optional[Union[PoolProxiedConnection, DBAPIConnection]],
494 cursor: Optional[DBAPICursor],
495 ) -> bool:
496 self.dbapi = cast("DBAPIModule", self.dbapi)
497 if isinstance(
498 e, self.dbapi.OperationalError
499 ) and "no active connection" in str(e):
500 return True
502 return super().is_disconnect(e, connection, cursor)
504 def get_driver_connection(
505 self, connection: DBAPIConnection
506 ) -> AsyncIODBAPIConnection:
507 return connection._connection # type: ignore[no-any-return]
509 def do_terminate(self, dbapi_connection: DBAPIConnection) -> None:
510 dbapi_connection.terminate()
513dialect = SQLiteDialect_aiosqlite