codekingpro/portable-devtools
114k
1# dialects/sqlite/aiosqlite.py
2# Copyright (C) 2005-2024 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# mypy: ignore-errors
8
9
10r"""
11
12.. dialect:: sqlite+aiosqlite
13 :name: aiosqlite
14 :dbapi: aiosqlite
15 :connectstring: sqlite+aiosqlite:///file_path
16 :url: https://pypi.org/project/aiosqlite/
17
18The aiosqlite dialect provides support for the SQLAlchemy asyncio interface
19running on top of pysqlite.
20
21aiosqlite is a wrapper around pysqlite that uses a background thread for
22each connection. It does not actually use non-blocking IO, as SQLite
23databases are not socket-based. However it does provide a working asyncio
24interface that's useful for testing and prototyping purposes.
25
26Using a special asyncio mediation layer, the aiosqlite dialect is usable
27as the backend for the :ref:`SQLAlchemy asyncio <asyncio_toplevel>`
28extension package.
29
30This dialect should normally be used only with the
31:func:`_asyncio.create_async_engine` engine creation function::
32
33 from sqlalchemy.ext.asyncio import create_async_engine
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
52Similarly to pysqlite, aiosqlite does not support SAVEPOINT feature.
53
54The solution is similar to :ref:`pysqlite_serializable`. This is achieved by the event listeners in async::
55
56 from sqlalchemy import create_engine, event
57 from sqlalchemy.ext.asyncio import create_async_engine
58
59 engine = create_async_engine("sqlite+aiosqlite:///myfile.db")
60
61 @event.listens_for(engine.sync_engine, "connect")
62 def do_connect(dbapi_connection, connection_record):
63 # disable aiosqlite's emitting of the BEGIN statement entirely.
64 # also stops it from emitting COMMIT before any DDL.
65 dbapi_connection.isolation_level = None
66
67 @event.listens_for(engine.sync_engine, "begin")
68 def do_begin(conn):
69 # emit our own BEGIN
70 conn.exec_driver_sql("BEGIN")
71
72.. warning:: When using the above recipe, it is advised to not use the
73 :paramref:`.Connection.execution_options.isolation_level` setting on
74 :class:`_engine.Connection` and :func:`_sa.create_engine`
75 with the SQLite driver,
76 as this function necessarily will also alter the ".isolation_level" setting.
77
78""" # noqa
79
80import asyncio
81from functools import partial
82
83from .base import SQLiteExecutionContext
84from .pysqlite import SQLiteDialect_pysqlite
85from ... import pool
86from ... import util
87from ...engine import AdaptedConnection
88from ...util.concurrency import await_fallback
89from ...util.concurrency import await_only
90
91
92class AsyncAdapt_aiosqlite_cursor:
93 # TODO: base on connectors/asyncio.py
94 # see #10415
95
96 __slots__ = (
97 "_adapt_connection",
98 "_connection",
99 "description",
100 "await_",
101 "_rows",
102 "arraysize",
103 "rowcount",
104 "lastrowid",
105 )
106
107 server_side = False
108
109 def __init__(self, adapt_connection):
110 self._adapt_connection = adapt_connection
111 self._connection = adapt_connection._connection
112 self.await_ = adapt_connection.await_
113 self.arraysize = 1
114 self.rowcount = -1
115 self.description = None
116 self._rows = []
117
118 def close(self):
119 self._rows[:] = []
120
121 def execute(self, operation, parameters=None):
122 try:
123 _cursor = self.await_(self._connection.cursor())
124
125 if parameters is None:
126 self.await_(_cursor.execute(operation))
127 else:
128 self.await_(_cursor.execute(operation, parameters))
129
130 if _cursor.description:
131 self.description = _cursor.description
132 self.lastrowid = self.rowcount = -1
133
134 if not self.server_side:
135 self._rows = self.await_(_cursor.fetchall())
136 else:
137 self.description = None
138 self.lastrowid = _cursor.lastrowid
139 self.rowcount = _cursor.rowcount
140
141 if not self.server_side:
142 self.await_(_cursor.close())
143 else:
144 self._cursor = _cursor
145 except Exception as error:
146 self._adapt_connection._handle_exception(error)
147
148 def executemany(self, operation, seq_of_parameters):
149 try:
150 _cursor = self.await_(self._connection.cursor())
151 self.await_(_cursor.executemany(operation, seq_of_parameters))
152 self.description = None
153 self.lastrowid = _cursor.lastrowid
154 self.rowcount = _cursor.rowcount
155 self.await_(_cursor.close())
156 except Exception as error:
157 self._adapt_connection._handle_exception(error)
158
159 def setinputsizes(self, *inputsizes):
160 pass
161
162 def __iter__(self):
163 while self._rows:
164 yield self._rows.pop(0)
165
166 def fetchone(self):
167 if self._rows:
168 return self._rows.pop(0)
169 else:
170 return None
171
172 def fetchmany(self, size=None):
173 if size is None:
174 size = self.arraysize
175
176 retval = self._rows[0:size]
177 self._rows[:] = self._rows[size:]
178 return retval
179
180 def fetchall(self):
181 retval = self._rows[:]
182 self._rows[:] = []
183 return retval
184
185
186class AsyncAdapt_aiosqlite_ss_cursor(AsyncAdapt_aiosqlite_cursor):
187 # TODO: base on connectors/asyncio.py
188 # see #10415
189 __slots__ = "_cursor"
190
191 server_side = True
192
193 def __init__(self, *arg, **kw):
194 super().__init__(*arg, **kw)
195 self._cursor = None
196
197 def close(self):
198 if self._cursor is not None:
199 self.await_(self._cursor.close())
200 self._cursor = None
201
202 def fetchone(self):
203 return self.await_(self._cursor.fetchone())
204
205 def fetchmany(self, size=None):
206 if size is None:
207 size = self.arraysize
208 return self.await_(self._cursor.fetchmany(size=size))
209
210 def fetchall(self):
211 return self.await_(self._cursor.fetchall())
212
213
214class AsyncAdapt_aiosqlite_connection(AdaptedConnection):
215 await_ = staticmethod(await_only)
216 __slots__ = ("dbapi",)
217
218 def __init__(self, dbapi, connection):
219 self.dbapi = dbapi
220 self._connection = connection
221
222 @property
223 def isolation_level(self):
224 return self._connection.isolation_level
225
226 @isolation_level.setter
227 def isolation_level(self, value):
228 # aiosqlite's isolation_level setter works outside the Thread
229 # that it's supposed to, necessitating setting check_same_thread=False.
230 # for improved stability, we instead invent our own awaitable version
231 # using aiosqlite's async queue directly.
232
233 def set_iso(connection, value):
234 connection.isolation_level = value
235
236 function = partial(set_iso, self._connection._conn, value)
237 future = asyncio.get_event_loop().create_future()
238
239 self._connection._tx.put_nowait((future, function))
240
241 try:
242 return self.await_(future)
243 except Exception as error:
244 self._handle_exception(error)
245
246 def create_function(self, *args, **kw):
247 try:
248 self.await_(self._connection.create_function(*args, **kw))
249 except Exception as error:
250 self._handle_exception(error)
251
252 def cursor(self, server_side=False):
253 if server_side:
254 return AsyncAdapt_aiosqlite_ss_cursor(self)
255 else:
256 return AsyncAdapt_aiosqlite_cursor(self)
257
258 def execute(self, *args, **kw):
259 return self.await_(self._connection.execute(*args, **kw))
260
261 def rollback(self):
262 try:
263 self.await_(self._connection.rollback())
264 except Exception as error:
265 self._handle_exception(error)
266
267 def commit(self):
268 try:
269 self.await_(self._connection.commit())
270 except Exception as error:
271 self._handle_exception(error)
272
273 def close(self):
274 try:
275 self.await_(self._connection.close())
276 except ValueError:
277 # this is undocumented for aiosqlite, that ValueError
278 # was raised if .close() was called more than once, which is
279 # both not customary for DBAPI and is also not a DBAPI.Error
280 # exception. This is now fixed in aiosqlite via my PR
281 # https://github.com/omnilib/aiosqlite/pull/238, so we can be
282 # assured this will not become some other kind of exception,
283 # since it doesn't raise anymore.
284
285 pass
286 except Exception as error:
287 self._handle_exception(error)
288
289 def _handle_exception(self, error):
290 if (
291 isinstance(error, ValueError)
292 and error.args[0] == "no active connection"
293 ):
294 raise self.dbapi.sqlite.OperationalError(
295 "no active connection"
296 ) from error
297 else:
298 raise error
299
300
301class AsyncAdaptFallback_aiosqlite_connection(AsyncAdapt_aiosqlite_connection):
302 __slots__ = ()
303
304 await_ = staticmethod(await_fallback)
305
306
307class AsyncAdapt_aiosqlite_dbapi:
308 def __init__(self, aiosqlite, sqlite):
309 self.aiosqlite = aiosqlite
310 self.sqlite = sqlite
311 self.paramstyle = "qmark"
312 self._init_dbapi_attributes()
313
314 def _init_dbapi_attributes(self):
315 for name in (
316 "DatabaseError",
317 "Error",
318 "IntegrityError",
319 "NotSupportedError",
320 "OperationalError",
321 "ProgrammingError",
322 "sqlite_version",
323 "sqlite_version_info",
324 ):
325 setattr(self, name, getattr(self.aiosqlite, name))
326
327 for name in ("PARSE_COLNAMES", "PARSE_DECLTYPES"):
328 setattr(self, name, getattr(self.sqlite, name))
329
330 for name in ("Binary",):
331 setattr(self, name, getattr(self.sqlite, name))
332
333 def connect(self, *arg, **kw):
334 async_fallback = kw.pop("async_fallback", False)
335
336 creator_fn = kw.pop("async_creator_fn", None)
337 if creator_fn:
338 connection = creator_fn(*arg, **kw)
339 else:
340 connection = self.aiosqlite.connect(*arg, **kw)
341 # it's a Thread. you'll thank us later
342 connection.daemon = True
343
344 if util.asbool(async_fallback):
345 return AsyncAdaptFallback_aiosqlite_connection(
346 self,
347 await_fallback(connection),
348 )
349 else:
350 return AsyncAdapt_aiosqlite_connection(
351 self,
352 await_only(connection),
353 )
354
355
356class SQLiteExecutionContext_aiosqlite(SQLiteExecutionContext):
357 def create_server_side_cursor(self):
358 return self._dbapi_connection.cursor(server_side=True)
359
360
361class SQLiteDialect_aiosqlite(SQLiteDialect_pysqlite):
362 driver = "aiosqlite"
363 supports_statement_cache = True
364
365 is_async = True
366
367 supports_server_side_cursors = True
368
369 execution_ctx_cls = SQLiteExecutionContext_aiosqlite
370
371 @classmethod
372 def import_dbapi(cls):
373 return AsyncAdapt_aiosqlite_dbapi(
374 __import__("aiosqlite"), __import__("sqlite3")
375 )
376
377 @classmethod
378 def get_pool_class(cls, url):
379 if cls._is_url_file_db(url):
380 return pool.NullPool
381 else:
382 return pool.StaticPool
383
384 def is_disconnect(self, e, connection, cursor):
385 if isinstance(
386 e, self.dbapi.OperationalError
387 ) and "no active connection" in str(e):
388 return True
389
390 return super().is_disconnect(e, connection, cursor)
391
392 def get_driver_connection(self, connection):
393 return connection._connection
394
395
396dialect = SQLiteDialect_aiosqlite
397